导读：一般情况下,我们不需要创建新的 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 个过程:

1. 自定义 Factory，根据需要实现 DynamicTableSourceFactory, DynamicTableSinkFactory.
2. 自定义 TableSink, 实现 DynamicTableSink
3. 定义 Options 也就是 connector 相关的属性
4. 在 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 {

    @Override
    public 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
     */
    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        return null;
    }

    @Override
    public String factoryIdentifier() {
        return "redis";
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        final Set<ConfigOption<?>> options = new HashSet();
        options.add(RedisOptions.HOST);
        options.add(RedisOptions.PORT);
        return options;
    }

    @Override
    public 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;
    }

    @Override
    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
        return ChangelogMode.newBuilder()
                .addContainedKind(RowKind.INSERT)
                .addContainedKind(RowKind.UPDATE_BEFORE)
                .addContainedKind(RowKind.UPDATE_AFTER)
                .build();
    }

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        // 获取 redis 的 host 和 port
        String 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);
    }

    @Override
    public DynamicTableSink copy() {
        return new RedisDynamicTableSink(this.options);
    }

    @Override
    public 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;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        this.jedis = new Jedis(host, port);
    }

    @Override
    public void invoke(RowData value, Context context) throws Exception {
        this.jedis.set(String.valueOf(value.getString(0)), String.valueOf(value.getInt(1)), "NX", "EX", expire);
    }

    @Override
    public 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 机制——ServiceLoader.load

使用&测试

到这里基本就完成了,下面来测试一下自定义的 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)
              
// 执行插入 SQL
              tEnv.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 了.

最后

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

原文地址：https://mp.weixin.qq.com/s/-h7fZxULS5cz2ZMgEEAncw