书接上回，上节说到了博主发现由于在 flink sql 中 lookup join 访问外部维表存在的性能问题。
由此诞生了一个想法，以 Redis 维表为例，Redis 支持 pipeline 批量访问模式，因此 flink sql lookup join 能不能按照 DataStream 方式一样，先攒一批数据 ，然后使用 Redis pipeline 批量访问外部存储。博主亲切的将这个功能称为 flink sql batch lookup join，本节就是讲述博主基于 flink 源码对此功能的实现。
废话不多说，咱们先直接上本文的目录和结论，小伙伴可以先看结论快速了解博主期望本文能给小伙伴们带来什么帮助：
来看看在具体场景下，对应输入值的输出值应该长啥样。
需求指标：使用曝光用户日志流（show_log）关联用户画像维表（user_profile）关联到用户的画像（性别，年龄段）数据。
来一波输入数据：
注意：redis 中的数据结构存储是按照 key，value 去存储的。其中 key 为 user_id，value 为 age，sex 的 json。如下图所示：
user_profile redis
预期输出数据如下：
batch lookup join sql 代码和原来的 lookup join sql 代码一模一样。如下 sql。
CREATE TABLE show_log (
log_id BIGINT,
`timestamp` as cast(CURRENT_TIMESTAMP as timestamp(3)),
user_id STRING,
proctime AS PROCTIME()
)
WITH (
'connector' = 'datagen',
'rows-per-second' = '10',
'fields.user_id.length' = '1',
'fields.log_id.min' = '1',
'fields.log_id.max' = '10'
);
CREATE TABLE user_profile (
user_id STRING,
age STRING,
sex STRING
) WITH (
'connector' = 'redis',
'hostname' = '127.0.0.1',
'port' = '6379',
'format' = 'json',
'lookup.cache.max-rows' = '500',
'lookup.cache.ttl' = '3600',
'lookup.max-retries' = '1'
);
CREATE TABLE sink_table (
log_id BIGINT,
`timestamp` TIMESTAMP(3),
user_id STRING,
proctime TIMESTAMP(3),
age STRING,
sex STRING
) WITH (
'connector' = 'print'
);
-- lookup join 的 query 逻辑
INSERT INTO sink_table
SELECT
s.log_id as log_id
, s.`timestamp` as `timestamp`
, s.user_id as user_id
, s.proctime as proctime
, u.sex as sex
, u.age as age
FROM show_log AS s
LEFT JOIN user_profile FOR SYSTEM_TIME AS OF s.proctime AS u
ON s.user_id = u.user_id
可以看到 lookup join 和 batch lookup join 的代码是完全相同的，唯一的不同之处在于，batch lookup join 需要设置 table config 参数，如下图所示：
table config
将原生 lookup join 和 batch lookup join 的效果做个对比：
原生的 lookup join：每输入一条数据，访问外部维表获取到结果输出一条数据，如下图所示。
lookup join
博主实现的 batch lookup join：是每攒够 30 条数据或者每 5s（防止数据量少的情况下，长时间不输出数据） 就利用 redis pipeline 能力访问外部存储一次。然后批量输出结果，如下图所示。大大提高了吞吐。
batch lookup join
博主将通过下面几个问题去交给大家怎么改源码去实现自己的功能。
在实现 batch lookup join 之前，当然要从原生的 lookup join 的实现开始入手，看看 flink 官方大大是怎么实现的，具体 transformation 如下图所示：
transformation
具体的实现逻辑承载在 org.apache.flink.streaming.api.operators.ProcessOperator，org.apache.flink.table.runtime.operators.join.lookup.LookupJoinRunner 中。
LookupJoinRunner 中的数据处理逻辑集中在 processElement 中。
LookupJoinRunner
可以看到上图，LookupJoinRunner 又内嵌了一层 fetcher 来实现具体的 lookup 逻辑。
接下来详细看看 fetcher 和 collector。
transformation fetcher
把这个 fetcher 的代码 copy 出来瞅瞅。
fetcher
fetcher 内嵌了 RedisRowDataLookupFunction 来作为最终访问外部维表的函数。
访问 redis 获取到数据。
RedisRowDataLookupFunction
transformation collector
把这个 collector 的代码 copy 出来瞅瞅。
collector
是不是感觉一个 lookup join 的调用链贼复杂。
因为 batch lookup join 是完全参考 lookup join 去实现的，所以接下来博主介绍一下整体的调用链关系，这就会方便后续设计 batch lookup join 实现方案的时候去确定具体修改哪一部分代码。
调用链
整体的调用逻辑如下：
还是一样，先看看设计思路最终的结论，batch lookup join 算子调用链设计如下：
batch lookup 调用链
详细说明一下设计思路：
思路有了，那么 batch lookup join 涉及到的改动项也就能确认了。
可以看到 is.dim.batch.mode 设置为 true 时，transformation 如下。transformation 中的重点处理逻辑就是 BatchLookupJoinRunner
batch transformation
BatchLookupJoinRunner
sql 生成的 fetcher 代码如下：
fetcher
RedisRowDataBatchLookupFunction 拿到输入的 List 数据，调用 Redis pipeline 批量访问外部存储。
RedisRowDataBatchLookupFunction
sql 生成的 collector 代码如下：
collector
目前上述方案实现的不足之处如下：
| 留言与评论（共有 0 条评论） “” |
