导读&＃xff1a;在FlinkSQL中TableSource提供对存储在外部系统(数据库&＃xff0c;键值存储&＃xff0c;消息队列)或文件中的数据的访问。 在TableEnvironment中注册TableSource后&＃xff0c;可以通过Table API或SQL查询对其进行访问。Flink 已内置了一批常用的Source & Sink&＃xff0c;如KafkaTableSource、JDBCTableSource等。本文将基于自定义KafkaTableSource从以下四点展开讨论&＃xff1a;

• 需要自定义KafkaTableSource场景描述
• 实现自定义KafkaTableSource的思路
• 自定义KafkaTableSource具体实现方式
• 自定义KafkaTableSourceFactory具体实现方式

场景描述

场景描述&＃xff1a;从多个渠道采集用户行为数据(数据来自不同渠道&＃xff0c;但是结构类型一致)&＃xff0c;并将采集到的数据传输到各个渠道在Kafka所对应的topic中&＃xff0c;之后由Flink Job访问Kafka获取数据并进行处理。

解决方案&＃xff1a;这里采用FlinkSQL进行Job开发&＃xff0c;通过使用内置的KafkaTableSource的connector.topic属性可以实现从指定Kafka topic获取消息。

bsTableEnv.sqlUpdate("CREATE TABLE sourceTable (" &＃43; "name STRING," &＃43; "country STRING" &＃43; ")" &＃43; "WITH (" &＃43; "&＃39;connector.type&＃39; &＃61; &＃39;kafka&＃39;," &＃43; "&＃39;connector.version&＃39; &＃61; &＃39;"&＃43;CONNECTOR_VERSION&＃43;"&＃39;," &＃43; "&＃39;connector.topic&＃39; &＃61; &＃39;"&＃43;SOURCE_TOPIC&＃43;"&＃39;," &＃43; "&＃39;connector.properties.zookeeper.connect&＃39; &＃61; &＃39;"&＃43;ZOOKEEPER_CONNECT&＃43;"&＃39;," &＃43; "&＃39;connector.properties.bootstrap.servers&＃39; &＃61; &＃39;"&＃43;BROKER_LIST&＃43;"&＃39;," &＃43; "&＃39;connector.properties.group.id&＃39; &＃61; &＃39;"&＃43;GROUP_ID&＃43;"&＃39;," &＃43; "&＃39;format.type&＃39; &＃61; &＃39;json&＃39;)");

美中不足之处&＃xff1a;由于数据虽然来自不同渠道&＃xff0c;但是结构类型一致。我们期望可以通过一个Job处理多个topic的消息&＃xff0c;而通过分析代码发现KafkaTableSource的connector.topic只能支持单个topic Name。代码如下&＃xff1a;

1、KafkaTableSource 通过 createKafkaConsumer 方法来创建 FlinkKafkaConsumer。

2、进入createKafkaConsumer发现其构造方法中对传入的String topic 通过Collections.singletonList(topic) 包装成了一个List。

基于以上分析发现KafkaTableSource的connector.topic并不能支持多个topic的写法。

实现自定义KafkaTableSource的思路

1、既然KafkaTableSource满足不了我们的需求&＃xff0c;而官方支持用户自定义Sources&＃xff0c;那么我们可以通过自定义一个KafkaTableSource来达到我们的需求&＃xff0c;Flink自定义Sources & Sink 文档地址&＃xff1a;

//FlinkSQL 1.10--User-defined Sources & Sinks https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/sourceSinks.html#define-a-tablesource

翻阅官方文档&＃xff0c;发现可以通过实现StreamTableSource接口来构建一个自定义StreamTableSource

2、接下来&＃xff0c;我们查看了StreamTableSource的层次结构&＃xff0c;发现Flink还提供了KafkaTableSourceBase类

StreamTableSource层次结构

KafkaTableSourceBase类实现了StreamTableSource接口&＃xff0c;同时也提供构建一个自定义KafkaTableSource所需的大部分基础内容(包括连接器connector的实现&＃xff0c;这里就不深入讨论&＃xff0c;感兴趣的朋友可以参考官方文档&＃xff1a;

https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/connectors/kafka.html)。

这里我们要关注一个点&＃xff0c;KafkaTableSourceBase类提供了一个createKafkaConsumer()的抽象方法&＃xff0c;用于创建KafkaConsumer。

KafkaTableSourceBase类

createKafkaConsumer

3、我们再看看KafkaTableSource类&＃xff0c;其继承了KafkaTableSourceBase类&＃xff0c;并实现了createKafkaConsumer()方法(这里要注意FlinkKafkaConsumer的版本问题)。

KafkaTableSource

4、深入查看FlinkKafkaConsumer类&＃xff0c;发现其提供了多个构造方法&＃xff0c;其中包括可接收List topics。

FlinkKafkaConsumer

那么到这里我们便可以参考KafkaTableSource例子&＃xff0c;通过继承KafkaTableSourceBase并根据自身需求实现createKafkaConsumer方法。

5、这里把类图简单画出来助于理解&＃xff0c;层次结构如下&＃xff1a;

自定义KafkaTableSource实现方式

1、编写一个MoreTopicKafkaTableSource类并继承 KafkaTableSourceBase&＃xff0c;然后重写createKafkaConsumer()方法。这里简单思路为将topic根据逗号进行切割并存成一个List。

MoreTopicKafkaTableSource

2、通过FlinkKafkaConsumer的 FlinkKafkaConsumer(List topics, DeserializationSchema deserializer, Properties props) 构造方法达到可以多topic的效果。

问题&＃xff1a;自此自定义的MoreTopicKafkaTableSource完成了&＃xff0c;但是如何将Source在环境中注册同时又如何让程序知道我们想用的是MoreTopicKafkaTableSource而不是内置的KafkaTableSource&＃xff1f;

自定义KafkaTableSourceFactory实现方式

1、查阅官方文档发现&＃xff0c;我们可以通过定义TableFactory来解决以上的问题。文档如下&＃xff1a;

TableFactory允许从基于字符串的属性中创建与表相关的不同实例。 调用所有可用的工厂以匹配给定的属性集和相应的工厂类。工厂利用Java的服务提供商接口(SPI)进行发现。 这意味着每个依赖项和JAR文件都应在META_INF / services资源目录中包含一个文件org.apache.flink.table.factories.TableFactory&＃xff0c;该文件列出了它提供的所有可用表工厂。

2、接下来&＃xff0c;我们参考KafkaTableSource类所对应的KafkaTableSourceSinkFactory工厂类&＃xff0c;该工厂类继承了KafkaTableSourceSinkFactoryBase。这里要关注的是KafkaTableSourceSinkFactory实现了createKafkaTableSource()方法&＃xff0c;在该方法中创建了KafkaTableSource对象。

3、同样&＃xff0c;我们也创建一个MoreTopicKafkaTableSourceSinkFactory并继承KafkaTableSourceSinkFactoryBase&＃xff0c;实现createKafkaTableSource()方法使其创建我们的MoreTopicKafkaTableSource对象。

4、在这里还有一个重要的点就是requiredContext()方法&＃xff0c;其用于匹配工厂。

指定为其实现此工厂的上下文。如果满足指定的属性和值集&＃xff0c;则框架保证仅与此工厂匹配。典型的属性可能是连接器类型, 格式.type&＃xff0c;或更新模式。属性键&＃xff0c;如connector.property-version 和 format.property-version后兼容情况保留的。

由于KafkaTableSourceSinkFactory和我们的自定义MoreTopicKafkaTableSourceSinkFactory都继承了KafkaTableSourceSinkFactoryBase&＃xff0c;而KafkaTableSourceSinkFactoryBase提供了requiredContext。这会导致框架匹配到两个工厂&＃xff0c;出现异常。

解决办法&＃xff1a;可以在MoreTopicKafkaTableSourceSinkFactory 自定义多一个key-value。

5、最后一个很重要的步骤是在 resource 目录下添加文件夹 META_INF/services&＃xff0c;并创建文件 org.apache.flink.table.factories.TableFactory&＃xff0c;在文件中写上新建的 Factory 类&＃xff0c;自此结束。

总结

1. 编写一个MoreTopicKafkaTableSource 继承 KafkaTableSourceBase&＃xff0c;并重写createKafkaConsumer方法。
2. 创建一个MoreTopicKafkaTableSourceSinkFactory并继承KafkaTableSourceSinkFactoryBase&＃xff0c;实现createKafkaTableSource()方法使其创建我们的MoreTopicKafkaTableSource。
3. 在MoreTopicKafkaTableSourceSinkFactory中重写requiredContext并为工厂匹配添加多一个key-value规则使匹配更加精确。
4. resource 目录下添加文件夹 META_INF/services&＃xff0c;并创建文件 org.apache.flink.table.factories.TableFactory&＃xff0c;在文件中写上新建的 Factory 类。

感谢您的阅读&＃xff0c;如果喜欢本文欢迎关注和转发&＃xff0c;本头条号将坚持持续分享IT技术知识。对于文章内容有其他想法或意见建议等&＃xff0c;欢迎提出共同讨论共同进步。