mysql 全量 同步 到 mysql调试一开始不知道为什么在哪里于是需要调试{ env{ execution.parallelism 4 job.mode BATCH } source { Jdbc { url jdbc:mysql://127.0.0.1:3306/demo username flink password 123456 driver com.mysql.cj.jdbc.Driver query SELECT * FROM test # 全量读取相关配置 fetch_size 1000 connection_pool_size 5 plugin_output fds_source } } sink { # 先输出到控制台验证数据 Console { plugin_input fds_source } LocalFile { plugin_input fds_source path /tmp/fds_to_oms_init_debug # 输出目录将在这里生成文件 file_format_type json # 推荐用 json结构清晰易读 custom_filename true # 是否自定义文件名 file_name_expression debug_output # 文件名前缀 is_enable_transaction false # 关闭事务就不要求 transactionId 了 batch_size 100 } } }控制台可以,有输出Job Statistic Information *********************************************** Start Time:2026-06-07 06:58:08 End Time:2026-06-07 06:58:10 Total Time(s):2Total Read Count:42Total Write Count:42Total Failed Count:0调试内容输出到文件了这说明问题在transfrom那么是什么原因导致的呢经过去掉transform中的查询条件发现transform {Sql{ plugin_inputfds_sourceplugin_outputoms_sinkquery select Id, DisplayName as UnitName, 3 as UnitType, 0 as TenantId from test where IsDeleted FALSE } }下面的方式就不行原因是因为数据库中定义的bit类型transform {Sql{ plugin_inputfds_sourceplugin_outputoms_sinkquery select Id, DisplayName as UnitName, 3 as UnitType, 0 as TenantId from test where IsDeleted 0 } }接着再调整sink入库方式,全量模式sink { Jdbc { plugin_input oms_sink url jdbc:mysql://127.0.0.1:3306/db2?useSSLfalseallowPublicKeyRetrievaltrueserverTimezoneAsia/Shanghai username ddd password 123456 driver com.mysql.cj.jdbc.Driver database db2 table test2 query insert into test2(Id,UnitName,UnitType,TenantId) values(?,?,?,?) } } }类型转换问题最简单的 case when 语法也不支持这个有点不像话吧。分析数据表字段,这个字段是 varchar类型因此需要类型转换FireControlJurisdictionIdvarchar(32)CHARACTERSETutf8COLLATEutf8_general_ciNULLDEFAULTNULL下面的转换还需要彻底一点因为将非数字都给筛掉了。 另外IsDeleted在这张表中是tinyint(1)类型也不能用0作为查询条件而是应该是1query SELECT Id, UnitName, IFNULL(UnitAddress, ) AS UnitAddress, IFNULL(ContactInformation, ) AS ContactTel, CASE WHEN CAST(FireControlJurisdictionId AS INT) 300 THEN 2 ELSE 1 END AS UnitType, FROM test22 WHERE IsDeletedfalse and FireControlJurisdiction AND FireControlJurisdictionId IS NOT NULL AND FireControlJurisdictionId AND FireControlJurisdictionId REGEXP ^[0-9]$ 升级seatunnel进入bin目录执行./install-plugin.sh下载seatunnel使用到的连接器。我是从2.3.12升级到2.3.13接着复制lib中的jar到2.3.13
mysql 全量 同步 到 mysql调试一开始不知道为什么在哪里于是需要调试{ env{ execution.parallelism 4 job.mode BATCH } source { Jdbc { url jdbc:mysql://127.0.0.1:3306/demo username flink password 123456 driver com.mysql.cj.jdbc.Driver query SELECT * FROM test # 全量读取相关配置 fetch_size 1000 connection_pool_size 5 plugin_output fds_source } } sink { # 先输出到控制台验证数据 Console { plugin_input fds_source } LocalFile { plugin_input fds_source path /tmp/fds_to_oms_init_debug # 输出目录将在这里生成文件 file_format_type json # 推荐用 json结构清晰易读 custom_filename true # 是否自定义文件名 file_name_expression debug_output # 文件名前缀 is_enable_transaction false # 关闭事务就不要求 transactionId 了 batch_size 100 } } }控制台可以,有输出Job Statistic Information *********************************************** Start Time:2026-06-07 06:58:08 End Time:2026-06-07 06:58:10 Total Time(s):2Total Read Count:42Total Write Count:42Total Failed Count:0调试内容输出到文件了这说明问题在transfrom那么是什么原因导致的呢经过去掉transform中的查询条件发现transform {Sql{ plugin_inputfds_sourceplugin_outputoms_sinkquery select Id, DisplayName as UnitName, 3 as UnitType, 0 as TenantId from test where IsDeleted FALSE } }下面的方式就不行原因是因为数据库中定义的bit类型transform {Sql{ plugin_inputfds_sourceplugin_outputoms_sinkquery select Id, DisplayName as UnitName, 3 as UnitType, 0 as TenantId from test where IsDeleted 0 } }接着再调整sink入库方式,全量模式sink { Jdbc { plugin_input oms_sink url jdbc:mysql://127.0.0.1:3306/db2?useSSLfalseallowPublicKeyRetrievaltrueserverTimezoneAsia/Shanghai username ddd password 123456 driver com.mysql.cj.jdbc.Driver database db2 table test2 query insert into test2(Id,UnitName,UnitType,TenantId) values(?,?,?,?) } } }类型转换问题最简单的 case when 语法也不支持这个有点不像话吧。分析数据表字段,这个字段是 varchar类型因此需要类型转换FireControlJurisdictionIdvarchar(32)CHARACTERSETutf8COLLATEutf8_general_ciNULLDEFAULTNULL下面的转换还需要彻底一点因为将非数字都给筛掉了。 另外IsDeleted在这张表中是tinyint(1)类型也不能用0作为查询条件而是应该是1query SELECT Id, UnitName, IFNULL(UnitAddress, ) AS UnitAddress, IFNULL(ContactInformation, ) AS ContactTel, CASE WHEN CAST(FireControlJurisdictionId AS INT) 300 THEN 2 ELSE 1 END AS UnitType, FROM test22 WHERE IsDeletedfalse and FireControlJurisdiction AND FireControlJurisdictionId IS NOT NULL AND FireControlJurisdictionId AND FireControlJurisdictionId REGEXP ^[0-9]$ 升级seatunnel进入bin目录执行./install-plugin.sh下载seatunnel使用到的连接器。我是从2.3.12升级到2.3.13接着复制lib中的jar到2.3.13
)
