标题: 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
简洁的步骤如下，记下来便于大家理解：
- 实现DynamicTableSinkFactory
- 实现DynamicTableSink
- 实现RichSinkFunction
- 配置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】
一起学习，一起进步。