flink sql 知其所以然（三）| 自定义 redis 数据汇表（附源码）

CREATE TABLE redis_sink_table (
    key STRING, -- redis key，第 1 列为 key
    `value` STRING -- redis value，第 2 列为 value
) WITH (
  'connector' = 'redis', -- 指定 connector 是 redis 类型的
  'hostname' = '127.0.0.1', -- redis server ip
  'port' = '6379', -- redis server 端口
  'write.mode' = 'string' -- 指定使用 redis `set key value`
)

INSERT INTO redis_sink_table
SELECT o.f0 as key, o.f1 as value
FROM leftTable AS o

public class RedisDynamicTableFactory implements DynamicTableSinkFactory {
    ...

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

    @Override
    public DynamicTableSink createDynamicTableSink(Context context) {

        // either implement your custom validation logic here ...
        // or use the provided helper utility
        final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);

        // validate all options
        // 所有 option 配置的校验，比如 write.mode 类参数
        helper.validate();

        // get the validated options
        final ReadableConfig options = helper.getOptions();

        final RedisWriteOptions redisWriteOptions = RedisOptions.getRedisWriteOptions(options);

        TableSchema schema = context.getCatalogTable().getSchema();

        / 创建 RedisDynamicTableSink
        return new RedisDynamicTableSink(schema.toPhysicalRowDataType()
                , redisWriteOptions);
    }
}

public class RedisDynamicTableSink implements DynamicTableSink {
    ...

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {

        // 初始化 redis 客户端配置
        FlinkJedisConfigBase flinkJedisConfigBase = new FlinkJedisPoolConfig.Builder()
                .setHost(this.redisWriteOptions.getHostname())
                .setPort(this.redisWriteOptions.getPort())
                .build();

        RedisMapper<RowData> redisMapper = null;

        switch (this.redisWriteOptions.getWriteMode()) {
            case "string":
                // redis key，value 序列化器
                // 从 RowData 转换成 redis 的 key value
                redisMapper = new SetRedisMapper();
                break;
            default:
                throw new RuntimeException("其他类型 write mode 请自定义实现");
        }

        // 创建 SinkFunction，注意！！！这里直接复用了 bahir 的实现
        return SinkFunctionProvider.of(new RedisSink<>(
                flinkJedisConfigBase
                , redisMapper));
    }
}

public class RedisRowDataLookupFunction extends TableFunction<RowData> {
    ...

    @Override
    public void invoke(IN input) throws Exception {
        String key = redisSinkMapper.getKeyFromData(input);
        String value = redisSinkMapper.getValueFromData(input);

        // 根据具体的命令执行具体写入 redis 的命令
        switch (redisCommand) {
            case RPUSH:
                this.redisCommandsContainer.rpush(key, value);
                break;
            case LPUSH:
                this.redisCommandsContainer.lpush(key, value);
                break;
            case SADD:
                this.redisCommandsContainer.sadd(key, value);
                break;
            case SET:
                this.redisCommandsContainer.set(key, value);
                break;
            case PFADD:
                this.redisCommandsContainer.pfadd(key, value);
                break;
            case PUBLISH:
                this.redisCommandsContainer.publish(key, value);
                break;
            case ZADD:
                this.redisCommandsContainer.zadd(this.additionalKey, value, key);
                break;
            case ZREM:
                this.redisCommandsContainer.zrem(this.additionalKey, key);
                break;
            case HSET:
                this.redisCommandsContainer.hset(this.additionalKey, key, value);
                break;
            default:
                throw new IllegalArgumentException("Cannot process such data type: " + redisCommand);
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        try {
            // 初始化 redis 执行器
            this.redisCommandsContainer = RedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);
            this.redisCommandsContainer.open();
        } catch (Exception e) {
            LOG.error("Redis has not been properly initialized: ", e);
            throw e;
        }
    }
}