FLINK实例(132):FLINK-SQL应用场景(23) CONNECTORS(23) 自定义 redis 数据维表(作为source表)（附源码）

CREATE TABLE dimTable (
    name StriNG,
    name1 StriNG,
    score BIGINT  -- redis 中存储数据的 scheR_441_11845@a
) WITH (
    'connector' = 'redis', -- 指定 connector 是 redis 类型的
    'hostname' = '127.0.0.1', -- redis server ip
    'port' = '6379', -- redis server 端口
    'format' = 'json' -- 指定 format 解析格式
    'lookup.cache.max-rows' = '500', -- guava local cache 最大条目
    'lookup.cache.ttl' = '3600', -- guava local cache ttl
    'lookup.max-retries' = '1' -- redis 命令执行失败后重复次数
)

SELECT o.f0, o.f1, c.name, c.name1, c.score
FROM leftTable AS o
-- 维表 join
LEFT JOIN dimTable FOR SYSTEM_TIME AS OF o.proctime AS c
ON o.f0 = c.nam

+I[a, b, namehhh, namehhh112, 3]
+I[a, b, namehhh, namehhh112, 3]
+I[a, b, namehhh, namehhh112, 3]
+I[a, b, namehhh, namehhh112, 3]
+I[a, b, namehhh, namehhh112, 3]
+I[a, b, namehhh, namehhh112, 3]
+I[a, b, namehhh, namehhh112, 3]
+I[a, b, namehhh, namehhh112, 3]
+I[a, b, namehhh, namehhh112, 3]

public class redisDynamicTableFactory implements DynamicTablesourceFactory {
    ...

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

    @Override
    public DynamicTablesource createDynamicTablesource(Context context) {

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

        // discover a suitable decoding format
        // format 实现
        final DecodingFormat<DeserializationscheR_441_11845@a<RowData>> decodingFormat = Helper.discoverDecodingFormat(
                DeserializationFormatFactory.class,
                FactoryUtil.FORMAT);

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

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

        final redisLookupOptions redisLookupOptions = redisOptions.getredisLookupOptions(options);

        TablescheR_441_11845@a scheR_441_11845@a = context.getCatalogTable().getscheR_441_11845@a();

        // 创建 redisDynamicTablesource
        return new redisDynamicTablesource(
                scheR_441_11845@a.toPhysicalRowDataType()
                , decodingFormat
                , redisLookupOptions);
    }
}

public class redisDynamicTablesource implements LookupTablesource {
    ...

    @Override
    public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {

        // 初始化 redis 客户端配置
        FlinkjedisConfigBase flinkjedisConfigBase = new FlinkjedisPoolConfig.builder()
                .setHost(this.redisLookupOptions.getHostname())
                .setPort(this.redisLookupOptions.getPort())
                .build();

        // redis key，value 序列化器
        LookupredisR_441_11845@apper lookupredisR_441_11845@apper = new LookupredisR_441_11845@apper(
                this.createDeserialization(context, this.decodingFormat, createValueFormatProjection(this.physicalDataTypE)));

        // 创建 table udf
        return TableFunctionProvider.of(new redisRowDataLookupFunction(
                flinkjedisConfigBase
                , lookupredisR_441_11845@apper
                , this.redisLookupOptions));
    }
}

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

    /**
     * 具体 redis 执行方法
     */
    public void eval(Object... objects) throws IOException {

        for (int retry = 0; retry <= maxRetryTimes; retry++) {
            try {
                // fetch result
                this.evaler.accept(objects);
                break;
            } catch (Exception E) {
                LOG.error(String.format("HBase lookup error, retry times = %d", retry), E);
                if (retry >= maxRetryTimes) {
                    throw new RuntimeException("Execution of redis lookup failed.", E);
                }
                try {
                    Thread.sleep(1000 * retry);
                } catch (InterruptedException e1) {
                    throw new RuntimeException(e1);
                }
            }
        }
    }


    @Override
    public void open(FunctionContext context) {
        LOG.info("start open ...");

        // redis 命令执行器，初始化 redis 链接
        try {
            this.redisCommandsContainer =
                    redisCommandsContainerBuilder
                            .build(this.flinkjedisConfigBasE);
            this.redisCommandsContainer.open();
        } catch (Exception E) {
            LOG.error("redis has not been properly initialized: ", E);
            throw new RuntimeException(E);
        }

        // 初始化 local cache
        this.cache = cacheMaxSize <= 0 || cacheExpireMs <= 0 ? null : CacheBuilder.newBuilder()
                .recordStats()
                .expireAfterWrite(cacheExpireMs, TimeUnit.MILLISECONDS)
                .maximumSize(cacheMaxSizE)
                .build();

        if (cache != null) {
            context.getMetricGroup()
                    .gauge("lookupCacheHitRate", (Gauge<Double>) () -> cache.stats().hitRate());


            this.evaler = in -> {
                RowData cacheRowData = cache.getIfPresent(in);
                if (cacheRowData != null) {
                    collect(cacheRowData);
                } else {
                    // fetch result
                    byte[] key = lookupredisR_441_11845@apper.serialize(in);

                    byte[] value = null;

                    switch (redisCommand) {
                        case GET:
                            value = this.redisCommandsContainer.get(key);
                            break;
                        case HGET:
                            value = this.redisCommandsContainer.hget(key, this.additionalKey.getBytes());
                            break;
                        default:
                            throw new IllegalArgumentexception("CAnnot process such data type: " + redisCommand);
                    }

                    RowData rowData = this.lookupredisR_441_11845@apper.deserialize(value);

                    collect(rowData);

                    cache.put(key, rowData);
                }
            };

        }
        ...
    }
}

public class redisDynamicTableFactory implements DynamicTablesourceFactory {
    ...
    @Override
    public DynamicTablesource createDynamicTablesource(Context context) {
        ...

        // discover a suitable decoding format
        // 复用 format 实现
        final DecodingFormat<DeserializationscheR_441_11845@a<RowData>> decodingFormat = Helper.discoverDecodingFormat(
                DeserializationFormatFactory.class,
                FactoryUtil.FORMAT);
        ...
    }
}

this.evaler = in -> {
    RowData cacheRowData = cache.getIfPresent(in);
    if (cacheRowData != null) {
        collect(cacheRowData);
    } else {
        // fetch result
        byte[] key = lookupredisR_441_11845@apper.serialize(in);

        byte[] value = null;

        switch (redisCommand) {
            case GET:
                value = this.redisCommandsContainer.get(key);
                break;
            case HGET:
                value = this.redisCommandsContainer.hget(key, this.additionalKey.getBytes());
                break;
            default:
                throw new IllegalArgumentexception("CAnnot process such data type: " + redisCommand);
        }

        RowData rowData = this.lookupredisR_441_11845@apper.deserialize(value);

        collect(rowData);

        cache.put(key, rowData);
    }
};