Flinkcdc研究

最近在研究Flinkcdc数据采集，底层技术为debezium，debezium会将日期转为5位数字，日期时间位13位的数字，看之前代码解决办法是：
1.识别十三位数字进行转换为日期格式。
2.对于date类型，人工穷举字段类型进行转换

我自己考虑了一下，这样可能会误转换、增大人工成本。感觉这样不是解决办法，就通过查看官网文档、看一些源码，看看是否有其他解决方法。

debezium自定义转换器

经过阅读debezium的官方文档，发现其是支持自定义转换器，因此可以通过自定义转换器时间事件类型的转换。
1.避免造轮子
经过搜索学习，发现github上有大佬已经写过针对mysql的时间点我直达转换器。
2.不得不造轮子
因为我日常参与的数据不仅是mysql、还有sqlserver。查看sqlserver并没有相关代码，就想着自己也写一个，能够兼容mysql和sqlserver的转换器（后来证明还是分开写比较好）。
3.学习分析
对大佬代码学习及jdbc源码查看，并且在实际的测试中。发现mysql、sqlserver的字段类型在快照、binlog(cdc)期间并不是完全一样。若要完全兼容，必须将映射类型找出来。
刚开始认为一个jdbc type只会对应一种java类。其实不是，而是在不同的jdbc中都有不同的映射。（通过chatgpt证实了一下）

mysql转换

mysql启动时，快照期间初始化转换器，在binlog期间仍进行一次初始化转换器。（使用的类不同）

字段类型	快照类型(jdbc type)	binlog类型(jdbc type)
DATE	java.time.LocalDate(93)	java.time.LocalDate(91)
TIME	java.time.Duration(92)	java.time.Duration(92)
DATETIME	java.sql.Timestamp(93)	java.time.LocalDateTime(93)
TIMESTAMP	java.sql.Timestamp(2014)	java.time.ZonedDateTime(2014)

sqlserver转换

sqlserver启动时快照期间初始化转换器，在cdc期间不再进行初始化转换器。（使用的类相同）

timestamp类型在sqlserver中为byte[]类型，jdbc type为-2，因此不进行转换。

字段类型	快照类型(jdbc type)	cdc类型(jdbc type)
DATE	java.sql.Date(91)	java.sql.Date(91)
TIME	java.sql.Timestamp(92)	java.sql.Time(92)
DATETIME	java.sql.Timestamp(93)	java.sql.Timestamp(93)
DATETIME2	java.sql.Timestamp(93)	java.sql.Timestamp(93)
DATETIMEOFFSET	microsoft.sql.DateTimeOffset(-155)	microsoft.sql.DateTimeOffset(-155)
SMALLDATETIME	java.sql.Timestamp(93)	java.sql.Timestamp(93)

开始写代码喽

java并不是我的擅长（在学了，），写代码期间借助了大量人工智能GitHub coplit，帮我实现。
刚开始涉及思路：大统一是全形式，将mysql和sqlserver都写到一个方法中去。

实际中发现并不是很合理。例如mysql的TIMESTAMP类型是时间戳，但sqlserver的TIMESTAMP是byte[]类型，还要在另外判断一下jdbc type是否为-2，也容易产生误解。最终决定分开写。

最后依照官网的模板重写方法就可以了。代码地址点击直达

使用方法

converters参数为：自定义转换器的名字，可以随意设置。设置的值就作为转换器的名字，在以后的参数中就要使用这个名字。
假设自定义的名字为mydebeziumconverter，则type参数为mydebeziumconverter.type。

mydebeziumconverter.type参数为：自定义转换器的类名，必须设置。（转换器的方法）
mydebeziumconverter.database.type参数为：数据库类型，必须设置。（需要设置为mysql或sqlserver)
mydebeziumconverter.format.datetime参数为：datetime类型的格式，可选。
mydebeziumconverter.format.date参数为：date类型的格式，可选。
mydebeziumconverter.format.time参数为：time类型的格式，可选。

如果仅使用mysql或sqlserver建议独立编译代码，只保留mysql或sqlserver的转换器，减少依赖。

flinkcdc

可使用源代码也可使用编译好的jar包。只需要放入目录即可。并在配置中设置参数。

// 自定义解释器
// 设置解释器及解释器参数
debeziumProperties.put("converters", "mydebeziumconverter");
debeziumProperties.put("mydebeziumconverter.type", "org.util.MyDebeziumConverter");
debeziumProperties.put("mydebeziumconverter.database.type", "mysql");
// 自定义格式，可选
debeziumProperties.put("mydebeziumconverter.format.datetime", "yyyy-MM-dd HH:mm:ss");
debeziumProperties.put("mydebeziumconverter.format.date", "yyyy-MM-dd");
debeziumProperties.put("mydebeziumconverter.format.time", "HH:mm:ss");

debezium

使用jar包，并将其放在 debezium 插件的同一级别目录中。并在配置文件中设置参数。

"converters": "mydebeziumconverter",
"mydebeziumconverter.type": "org.util.MyDebeziumConverter",
"mydebeziumconverter.database.type": "mysql",
"mydebeziumconverter.format.datetime": "yyyy-MM-dd HH:mm:ss",
"mydebeziumconverter.format.date": "yyyy-MM-dd",
"mydebeziumconverter.format.time": "HH:mm:ss"

github

代码已发布到github，点击直达

FlinkCDC/DEBEZIUM自定义日期转换器