Flink实战之扩展JDBC Dialect

前言

Flink jdbc connect已经为依赖于jdbc进行操作的数据库系统提供了非常完善的功能。对于扩展只需要扩展其Dialect即可。

具体步骤

1. 定义CustomizedDialect类，具体可参考MySQLDialect
   修改点：
   • TIMESTAMP_PRECISION
   • DECIMAL_PRECISION
   • dialectName()
   • canHandle() → "jdbc:xxx:" 如："jdbc:clickhouse:"
   • getRowConverter() → 定义CustomizedRowConverter，参考MySQLRowConverter
   • defaultDriverName() 驱动名称 如："ru.yandex.clickhouse.ClickHouseDriver"
   • getUpsertStatement() 根据是否需要主键更新而定
2. 将CustomizedDialect添加到JdbcDialects的dialect列表中

经过上面两步一个简单的dialect就实现了，jdbc-connector是通过database的url来判断具体选择哪种方言来执行的。

详细代码：https://github.com/zhuxiaoshang/flink-be-god/tree/master/flink-connector/flink-jdbc-dialect-customized/src/main/java/jdbc/dialect/customized