Apache Flink 漫谈系列 - 持续查询(Continuous Queries)

上图描述了一个双流JOIN的场景，双流JOIN的底层实现会将左(L)右(R)两面的数据都持久化到Apache Flink的State中，当L流入一条事件，首先会持久化到LState，然后在和RState中存储的R中所有事件进行条件匹配，这样的逻辑如果R流product_id为P001的产品销售记录已经流入4条，L流的(P001, 48) 流入的时候会匹配4条事件流入下游(join_sink)。

2. 问题

上面双流JOIN的场景，我们发现其实inventory和sales表是有业务的PK的，也就是两张表上面的product_id是唯一的，但是由于我们在Sorure上面无法定义PK字段，表上面所有的数据都会以append only的方式从source流入到下游计算节点JOIN，这样就导致了JOIN内部所有product_id相同的记录都会被匹配流入下游，上面的例子是 (P001, 48) 来到的时候，就向下游流入了4条记录，不难想象每个product_id相同的记录都会与历史上所有事件进行匹配，进而操作下游数据压力。

那么这样的压力是必要的吗?从业务的角度看，不是必要的，因为对于product_id相同的记录，我们只需要对左右两边最新的记录进行JOIN匹配就可以了。比如(P001, 48)到来了，业务上面只需要右流的(P001, 22)匹配就好，流入下游一条事件(P001, 48, 22)。 那么目前在Apache Flink上面如何做到这样的优化呢?

3. 解决方案

上面的问题根本上我们要构建一张有PK的动态表，这样按照业务PK进行更新处理，我们可以在Source后面添加group by 操作生产一张有PK的动态表。如下：

SQL: 
 
CREATE TABLE inventory_tab( 
product_id VARCHAR, 
product_count BIGINT 
) 
 
CREATE TABLE sales_tab( 
product_id VARCHAR, 
sales_count BIGINT 
) 
CREATE VIEW inventory_view AS 
SELECT 
product_id, 
LAST_VALUE(product_count) AS product_count 
FROM inventory_tab 
GROUP BY product_id; 
 
CREATE VIEW sales_view AS 
SELECT 
product_id, 
LAST_VALUE(sales_count) AS sales_count 
FROM sales_tab 
GROUP BY product_id; 
 
CREATE TABLE join_sink( 
product_id VARCHAR, 
product_count BIGINT, 
sales_count BIGINT, 
PRIMARY KEY(product_id) 
)WITH ( 
type = 'print' 
) ; 
 
CREATE VIEW join_view AS 
SELECT 
l.product_id, 
l.product_count, 
r.sales_count 
FROM inventory_view l 
JOIN sales_view r 
ON l.product_id = r.product_id; 
 
INSERT INTO join_sink 
SELECT 
product_id, 
product_count, 
sales_count 
FROM join_view ; 

（编辑：青岛站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!