如何同步mysql数据到Doris中

Doris官网定义 mysql原始表结构 1.doris中关联mysql外表 结果如下： 2.doris中关联kafka导入数据 查看作业 State为RUNNING，表示已经成功。 停止作业 3.通过flink导入mysql数据到doris 方法1：通过mysql-cdc写入kafka，kafka关联doris表。 方法2：通过阿里云DTS->datahub，然后通过Flink写入kafka，再关联到doris外表 如何处理delete数据？对于方法1，需要手动的删除doris中的数据；对于方法2，可以通过dts_operation_flag字段来标示，dts_operation_flag可以为I/U/D，分别表示添加、更新和删除。那我们就只需要在doris表中添加一个dts_operation_flag字段来标示就可以了，查询数据的时候就不再查询等于D的值。 如何处理脏数据？delete doris中的数据，然后insert正确的值；还有个方法是将关联一个外表（这个是正确的值），然后再将doris中的表和外表中的值diff，将diff的值insert到doris中。

目前业务上有同步数据到sqlServer的需求，但是flink1.12.1版本的JdbcDialects不支持SqlServerDialect,

科学上网后发现袋鼠云的flinkStreamSql已经有支持sqlserver，那就开始动手,参考实现一波

主要实现getUpsertStatement的方法，本来以为能直接copy一波flinkStreamSql 的实现，结果发现

报错 SQL statement must not contain ? character.

查看源码发现, flink在构建mysql的Statement,是先把需要替换的字段前面拼接了 : fieldNames，然后在org.apache.flink.connector.jdbc.statement.FieldNamedPreparedStatementImpl类的parseNamedStatement 替换成 ?号， 既然如此，就针对了buildDualQueryStatement进行修改

完整的SqlServerDialect文件

最后替换原有的flink-jar包后，就可以用类似flink定义mysql的ddl进行定义表了

注意url写法为：jdbc:jtds:sqlserver://xxx:1433databaseName=master

[flinkStreamSQL链接] https://github.com/DTStack/flinkStreamSQL/blob/1.11_release/sqlserver/sqlserver-sink/src/main/java/com/dtstack/flink/sql/sink/sqlserver/SqlserverDialect.java

---