Flink SQL 自定义 redis connector

一般情况下,我们不需要创建新的 connector,因为 Flink SQL 已经内置了丰富的 connector 供我们使用,但是在实际生产环境中我们的存储是多种多样的,所以原生的 connector 并不能满足所有用户的需求,这个时候就需要我们自定义 connector,这篇文章的重点就是介绍一下如何实现自定义 Flink SQL connector ?

先来看一下官网的一张 connector 架构图:

Metadata

Table API 和 SQL 都是声明式 API。这包括表的声明。因此，执行 CREATE TABLE 语句会导致目标目录中的元数据更新。对于大多数目录实现，不会为此类操作修改外部系统中的物理数据。特定于连接器的依赖项不必出现在类路径中。WITH 子句中声明的选项既不经过验证也不以其他方式解释。动态表的元数据（通过 DDL 创建或由目录提供）表示为 CatalogTable 的实例。必要时，表名将在内部解析为 CatalogTable。

Planning

在规划和优化表程序时，需要将 CatalogTable 解析为 DynamicTableSource（用于在 SELECT 查询中读取）和 DynamicTableSink（用于在 INSERT INTO 语句中写入）。

DynamicTableSourceFactory 和 DynamicTableSinkFactory 提供特定于连接器的逻辑，用于将 CatalogTable 的元数据转换为 DynamicTableSource 和 DynamicTableSink 的实例。在大多数情况下，工厂的目的是验证选项（例如示例中的 'port' = '5022'）、配置编码/解码格式（如果需要）以及创建表连接器的参数化实例。

默认情况下，使用 Java 的服务提供者接口 (SPI) 发现 DynamicTableSourceFactory 和 DynamicTableSinkFactory 的实例。连接器选项（例如示例中的 'connector' = 'custom'）必须对应于有效的工厂标识符。

尽管在类命名中可能不明显，但 DynamicTableSource 和 DynamicTableSink 也可以被视为有状态的工厂，它们最终为读取/写入实际数据生成具体的运行时实现。

规划器使用源和接收器实例来执行特定于连接器的双向通信，直到找到最佳逻辑计划。根据可选声明的能力接口（例如 SupportsProjectionPushDown 或 SupportsOverwrite），规划器可能会对实例应用更改，从而改变生成的运行时实现。

Runtime

一旦逻辑规划完成，规划器将从表连接器获取运行时实现。运行时逻辑在 Flink 的核心连接器接口中实现，例如 InputFormat 或 SourceFunction。

这些接口按另一个抽象级别分组为 ScanRuntimeProvider、LookupRuntimeProvider 和 SinkRuntimeProvider 的子类。

例如，OutputFormatProvider（提供 org.apache.flink.api.common.io.OutputFormat）和 SinkFunctionProvider（提供 org.apache.flink.streaming.api.functions.sink.SinkFunction）都是 SinkRuntimeProvider 的具体实例，规划器可以处理。

自定义 redis sink connector

大概需要下面 4 个过程:

自定义 Factory，根据需要实现 DynamicTableSourceFactory, DynamicTableSinkFactory.
自定义 TableSink, 实现 DynamicTableSink
定义 Options 也就是 connector 相关的属性
在 resource 下面添加配置文件 org.apache.flink.table.factories.Factory 里面添加 Factory 的全限定名

Factory

package flink.connector.redis;import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableSinkFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
import java.util.HashSet;
import java.util.Set;/*** 自定义 Factory **/
public class RedisDynamicTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {@Overridepublic DynamicTableSink createDynamicTableSink(Context context) {final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);helper.validate();ReadableConfig options = helper.getOptions();return new RedisDynamicTableSink(options);}/*** 这里没有实现 source * @param context* @return*/@Overridepublic DynamicTableSource createDynamicTableSource(Context context) {return null;}@Overridepublic String factoryIdentifier() {return "redis";}@Overridepublic Set<ConfigOption<?>> requiredOptions() {final Set<ConfigOption<?>> options = new HashSet();options.add(RedisOptions.HOST);options.add(RedisOptions.PORT);return options;}@Overridepublic Set<ConfigOption<?>> optionalOptions() {final Set<ConfigOption<?>> options = new HashSet();options.add(RedisOptions.EXPIRE);return options;}
}

TableSink

package flink.connector.redis;import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.SinkFunctionProvider;
import org.apache.flink.types.RowKind;
import static flink.connector.redis.RedisOptions.*;/*** 自定义 DynamicTableSink **/
public class RedisDynamicTableSink implements DynamicTableSink {private ReadableConfig options;public RedisDynamicTableSink(ReadableConfig options) {this.options = options;}@Overridepublic ChangelogMode getChangelogMode(ChangelogMode requestedMode) {return ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT).addContainedKind(RowKind.UPDATE_BEFORE).addContainedKind(RowKind.UPDATE_AFTER).build();}@Overridepublic SinkRuntimeProvider getSinkRuntimeProvider(Context context) {// 获取 redis 的 host 和 portString host = options.getOptional(HOST).get();Integer port = options.getOptional(PORT).get();Integer expire = options.getOptional(EXPIRE).get();MyRedisSink myRedisSink = new MyRedisSink(host, port, expire);return SinkFunctionProvider.of(myRedisSink);}@Overridepublic DynamicTableSink copy() {return new RedisDynamicTableSink(this.options);}@Overridepublic String asSummaryString() {return "redis table sink";}
}

package flink.connector.redis;import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.table.data.RowData;
import redis.clients.jedis.Jedis;/*** 自定义 sink 写入 redis**/
public class MyRedisSink extends RichSinkFunction<RowData> {private final String host;private final int port;private int expire;private Jedis jedis;public MyRedisSink(String host, int port, int expire) {this.host = host;this.port = port;this.expire = expire;}@Overridepublic void open(Configuration parameters) throws Exception {this.jedis = new Jedis(host, port);}@Overridepublic void invoke(RowData value, Context context) throws Exception {this.jedis.set(String.valueOf(value.getString(0)), String.valueOf(value.getInt(1)), "NX", "EX", expire);}@Overridepublic void close() throws Exception {this.jedis.close();}
}

这里用的是 Jedis 没有使用 Flink 自带的 redis connector ,因为 Flink 自带的功能有限,很多功能都需要自己扩展,所以就直接使用 Jedis.我这里只是为了演示,只实现了最简单的 set 功能.

Options

package flink.connector.redis;import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;/** * Option utils for redis table source sink. */
public class RedisOptions {private RedisOptions() {}public static final ConfigOption<String> HOST =ConfigOptions.key("host").stringType().noDefaultValue().withDescription("The Redis table host.");public static final ConfigOption<Integer> PORT =ConfigOptions.key("port").intType().defaultValue(6379).withDescription("The Redis table port.");public static final ConfigOption<Integer> EXPIRE =ConfigOptions.key("expire").intType().noDefaultValue().withDescription("The Redis table expire time.");
}

所有 redis 相关的属性都可以在这里添加,比如用户名密码,连接池相关的配置等.

配置文件

最后也是最重要的一点就是在 resource 下面添加配置文件,因为 Flink 是通过 SPI 机制来发现工厂的,如果有不了解 SPI 机制的可以看前面这篇文章,注意这个路径一定不要写错.

到这里基本就完成了,下面来测试一下自定义的 connector 能否把数据准确的写入到 redis 里面.

使用&测试

// 定义数据源表
tEnv.executeSql("""|CREATE TABLE datagen (| f_sequence INT,| f_random INT,| f_random_str STRING,| ts AS localtimestamp,| WATERMARK FOR ts AS ts|) WITH (| 'connector' = 'datagen',| -- optional options --| 'rows-per-second'='1',| 'fields.f_sequence.kind'='sequence',| 'fields.f_sequence.start'='1',| 'fields.f_sequence.end'='20',| 'fields.f_random.min'='1',| 'fields.f_random.max'='1000',| 'fields.f_random_str.length'='10'|)|""".stripMargin)// 定义 redis 表tEnv.executeSql("""|create table redis_sink (|f1 STRING,|f2 INT|) WITH (|'connector' = 'redis',|'host' = 'xxx',|'port' = '6379',|'expire' = '100'|)|""".stripMargin)// 执行插入 SQLtEnv.executeSql("""|insert into redis_sink|select f_random_str,f_random|from datagen|""".stripMargin)

上面的 datagen 会产生 20 条数据.执行上面的 SQL 然后查询一下 redis 打印的数据如下:

68652c3a52 : 396
de3044d6d0 : 248
b09690ec10 : 436
dab4bb9ea9 : 821
d57a47d883 : 134
4d3d23767a : 63
9ca712a25f : 527
cb3019326d : 164
4a4af63f89 : 803
3cb960dbf1 : 575
db95bf7590 : 500
4274665b4b : 910
5c27396cb1 : 993
c1d957a2c8 : 951
8b24d7abe2 : 66
817b59d742 : 354
baa51bb58a : 14
db32f9cd53 : 510
3c5db2220b : 44
7c169eaef9 : 160

通过上面的 Demo,相信大家对自定义 Flink SQL connector 已经有所了解,那在生产环境中就可以根据自己的需求去定制各种 connector 了.