FlinkSQL 之 temporary join 开发
关于实时flinkSQL的双流join的背景知识可以先阅读以下文章：
https://www.51cto.com/article/713922.html
目前我们有一条流量日志明细的TT流A，以及一条商品标签的TT流B，在flink中对A流和B流进行双流join类似于将A流关联一个hbase维表。temporary join有以下特点：
1. 单流驱动：虽然是双流join，但数据下发只由一条流驱动。
2. 需要定义versioned table，versioned table记录了每个时刻的属性信息，双流join时被动查询。类似于银行汇率表，在货币兑换的时候需要参考兑换时刻的汇率。
3. 查询携带时间版本信息：temporary join携带由两条流的watermark触发，因此查询到的属性是对应时间内的属性。
![]()
应用场景&实例分享
当需要根据实时汇率*货币金额计算总金额，实时商品价格*成交件数计算总成交金额时，经常会使用temporary join获取实时的汇率和价格信息。在笔者的流量升级业务迭代中，我们需要获取实时的商品标签，因此需要定义商品标签的versioned table，写法如下：
CREATE TEMPORARY TABLE `tag_ri` (`id` VARCHAR,`tag` VARCHAR,`time` VARCHAR,`ts` AS `TO_TIMESTAMP`(`time`, 'yyyy-MM-dd HH:mm:ss'),WATERMARK FOR `ts` AS `withOffset`(`ts`, 0) --定义watermark) WITH ('connector' = 'tt','router' = '******','topic' = 'tag_ri','lineDelimiter' = '\n','fieldDelimiter' = '\u0001','encoding' = 'utf-8');--定义version tableCREATE TEMPORARY VIEW `tag`ASSELECT `id`, `tag`, `time`, `ts`FROM ( SELECT `id`, `tag`, `time`, `ts`, ROW_NUMBER() OVER (PARTITION BY `id` --关联主键ORDER BY `time` DESC) AS `rownum`FROM `tag_ri`)WHERE `rownum` = 1;
同上我们也需要定义流量日志明细流的watermark，并进行双流join
CREATE TEMPORARY TABLE `log_ri` (`id` VARCHAR,`time` VARCHAR,......`ts` AS `TO_TIMESTAMP`(`time`, 'yyyy-MM-dd HH:mm:ss'),WATERMARK FOR `ts` AS `withOffset`(`ts`, 0)) WITH ('connector' = 'tt','router' = '******','topic' = 'log_ri','lineDelimiter' = '\n','fieldDelimiter' = '\u0001','encoding' = 'utf-8',);select `a`.`id`,......,`b`.`tag`from (SELECT *FROM `log_ri`) AS `a`LEFT JOIN `tag` FOR SYSTEM_TIME AS OF `a`.`ts` AS `b` ON `a`.`id` = `b`.`id`
结果如下：
--商品标签信息12:00> SELECT * FROM tag_ri;id tag(商品标签)======= =======================t1 A12:30> SELECT * FROM tag_ri;id tag(商品标签)======= =======================t1 B--流量明细日志查询 t1商品共三条明细SELECT * FROM log_ri;id time======= ========t1 12:00t1 12:15t1 12:30--执行temporary joinselect `a`.`id`,`a`.`time`,`b`.`tag`from (SELECT *FROM `log_ri`) AS `a`LEFT JOIN `tag` FOR SYSTEM_TIME AS OF `a`.`ts` AS `b` ON `a`.`id` = `b`.`id`id time tag(商品标签)======= ======== =======================t1 12:00 At1 12:15 At1 12:30 B
![]()
开发经验
▐ 稀疏数据处理
由于temporary join是由两条流的watermark触发，如果versioned table是一条稀疏的流（在一段时间内无数据流入），那么join可能存在等待不下发数据的现象，可以通过设置参数 set table.exec.source.idle-timeout = 10s ，可以让A流数据不进行等待，具体参数介绍可以参考：
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/config/#table-exec-source-idle-timeout
▐ 数据延迟下发
-
问题
在实际开发中，我们发现temporay join后数据一直等待不下发，整点才会进行下发的现象。
-
原因分析
我们结合SQL语法，对TT日志进行回流分析：代码逻辑是四路source union后， join 定义的versioned table
select a.*,b.tagfrom(select * from source_1union allselect * from source_2union allselect * from source_3union allselect * from source_4) atemporay joinb流
-
解法
总结
-
https://www.51cto.com/article/713922.html
-
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/config/#table-exec-source-idle-timeout
团队介绍
我们是淘天集团-业务技术-商家数据团队，专注于开发和维护生意参谋这一全渠道、全链路、一站式的数据平台，同时也负责品牌数据银行和策略中心两大产品。旨在为商家提供全面的数据服务，包括但不限于经营分析、市场洞察、客群洞察等，以帮助商家提高商业决策效率。
本文分享自微信公众号 - 大淘宝技术（AlibabaMTT）。
如有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一起分享。
