标题: Flink sql自定义connector
日期: 2021-09-26 22:09:54
标签: [Flink, connector]
分类: Flink

最近公司正在做实时数仓相关的东西，我呢，负责实施，市面上的实时数仓的建设文章很多，发现大部分是基于flink的，国内的朋友，尤其是阿里，力推flink，那么我也来尝试下，使用flink作为我们实时数仓的计算引擎。

第一步就是调研flink如何接入数据，在这一块，我们就遇到了麻烦，因为我们的数据源大部分是oracle，flink官方目前还未提供oracle-cdc（现在用的是Flink 1.12.1），所以怎么办呢，自己定义Flink oracle的cdc工具jar包，ok，这一步搞定了。

原本我们计划架构是这样走：

mysql/oracle  ->  flink cdc -> hbase  ->  hive
              ->  kafka     -> 

这样我们的实时数据，你在hive中也能够查询，多好，但是目前Flink只支持hbase的1.4和2.2版本，我们公司是通过hbase1.0建设起来的，而且经过多番沟通，组件无法升级，因为还包含了许多自定义的内容在里面，如果升级，那么这些东西需要重新写，比较麻烦，于是我们决定自己写hbase sink，实现flink sql到hbase的写入。

那么，如何自定义hbase sink呢，官方提供了样例，我们来看看：
Flink Self-defined Sources and Sinks
简洁的步骤如下，记下来便于大家理解：

1. 实现DynamicTableSinkFactory
2. 实现DynamicTableSink
3. 实现RichSinkFunction
4. 配置Java SPI

第一步：
实现DynamicTableSinkFactory接口，需要实现方法createDynamicTableSink，返回DynamicTableSink对象
那么就到第二步了。
其他的方法requiredOptions、optionalOptions都是你使用sql时，必需和可选的参数

第二步：
实现DynamicTableSink接口，需要实现方法：getSinkRuntimeProvider，这个方法就是sql与streaming建立联系的地方
需要调用你的sink逻辑方法function。

第三步：
实现RichSinkFunction抽象类，需要实现方法：invoke，这个方法就是你实现sink逻辑的地方。

第四步：
这一步，也很关键，因为，你要让flink程序能够找到你自定义的connector，就必需指定到固定位置
在resources目录下建文件夹META-INF/services，并创建你的factory文件：

META-INF/services/org.apache.flink.table.factories.Factory

文件中指定你实现的DynamicTableSinkFactory类即可。

好了，祝大家体验成功。
有什么问题，可以联系我：
chenzuoli709@163.com
wx:15313621879

书山有路勤为径，学海无涯苦作舟。

欢迎关注我的微信公众号，比较喜欢分享知识，也喜欢宠物，所以做了这2个公众号：

喜欢宠物的朋友可以关注：【电巴克宠物Pets】

一起学习，一起进步。