Flink SQL多字段的维表关联查询-flink-connec

Flink SQL经常需要关联维表，使用redis作为维表存储介质能获得较快的响应速度，大多情况维表有多个字段,写入与维表查询时只能使用concat与table function函数处理，非常麻烦。有没有更简单的方法呢？

本示例基于flink-connector-redis展示如何利用'value.data.structure'='row'写多字段并关联查询，无需定义table function函数。

写入多字段的维表测试数据

-- 创建表
create table sink_redis(uid VARCHAR,score double,score2 double )
with ( 'connector' = 'redis',
            'host' = '10.11.69.176',
            'port' = '6379',
            'redis-mode' = 'single',
            'password' = '****',
            'command' = 'SET',
            'value.data.structure' = 'row');  -- 'value.data.structure'='row':整行内容保存至value并以'\01'分割
-- 写入测试数据，score、score2为需要被关联查询出的两个维度
insert into sink_redis select * from (values ('1', 10.3, 10.1));

写入数据后，在redis中value保存的value为: "1\x0110.3\x0110.1",至此关联数据准备就绪。

创建源表与关联表

-- create join table --
create table join_table with ('command'='get', 'value.data.structure'='row') like sink_redis

-- create result table --
create table result_table(uid VARCHAR, username VARCHAR, score double, score2 double) with ('connector'='print')

-- create source table --
create table source_table(uid VARCHAR, username VARCHAR, proc_time as procTime()) with ('connector'='datagen', 'fields.uid.kind'='sequence', 'fields.uid.start'='1', 'fields.uid.end'='2')

关联查询

insert
    into
    result_table
select
    s.uid,
    s.username,
    j.score, -- 来自维表
    j.score2 -- 来自维表
from
    source_table as s
join join_table for system_time as of s.proc_time as j on
    j.uid = s.uid

查询结果:

2> +I[2, 1e0fe885a2990edd7f13dd0b81f923713182d5c559b21eff6bda3960cba8df27c69a3c0f26466efaface8976a2e16d9f68b3, null, null]
1> +I[1, 30182e00eca2bff6e00a2d5331e8857a087792918c4379155b635a3cf42a53a1b8f3be7feb00b0c63c556641423be5537476, 10.3, 10.1]

项目地址：

插件地址：https://github.com/jeff-zou/flink-connector-redis.git

无法翻墙：https://gitee.com/jeff-zou/flink-connector-redis.git

使用方法:

在命令行执行 mvn package -DskipTests打包后，将生成的包flink-connector-redis-1.1.0.jar引入flink lib中即可，无需其它设置。
项目依赖jedis 3.7.1,如flink环境无jedis,则使用flink-connector-redis-1.1.0-jar-with-dependencies.jar。
开发环境工程直接引用：

<dependency>
    <groupId>io.github.jeff-zou</groupId>
    <artifactId>flink-connector-redis</artifactId>
    <version>1.1.0</version>
</dependency>