FlinkSQL 之乱序问题

乱序问题

在业务编写 FlinkSQL 时， 非常常见的就是乱序相关问题， 在出现问题时，非常难以排查，且无法稳定复现，这样无论是业务方，还是平台方，都处于一种非常尴尬的地步。
在实时 join 中， 如果是 Regular Join， 则使用的是 Hash Join 方式， 左表和右表根据 Join Key 进行hash，保证具有相同 Join Key 的数据能够 Hash 到同一个并发，进行 join 的计算 。
在实时聚合中， 主要普通的 group window， over window， time window 这几中开窗方式，都涉及到 task 和 task 之间 hash 方式进行数据传输。
因此， 在比较复杂的逻辑中， 一条数据在整个数据流中需要进行不同的 hash 方式， 特别时当我们处理 CDC 数据时， 一定要要求数据严格有序， 否则可能会导致产生错误的结果。
以下面的例子进行说明， 以下有三张表， 分别是订单表， 订单明细表， 和商品类目 。

• 这三张表的实时数据都从 MySQL 采集得到并实时写入 Kafka， 均会实时发生变化， 无法使用窗口计算
  
• 除了订单表有订单时间， 其他两张表都没有时间属性， 因此无法使用watermark
  

1. CREATE TABLE orders (
2.         order_id VARCHAR,
3.         order_time TIMESTAMP
4. ) WITH (
5.         'connector' = 'kafka',
6.         'format' = 'changelog-json'
7.         ...
8. );
10. CREATE TABLE order_item (
11.         order_id VARCHAR,
12.         item_id VARCHAR
13. ) WITH (
14.         'connector' = 'kafka',)
15.         'format' = 'changelog-json'
16.         ...
17. );
19. CREATE TABLE item_detail (
20.         item_id VARCHAR,
21.         item_name VARCHAR,
22.         item_price BIGINT
23. ) WITH (
24.         'connector' = 'kafka',
25.         'format' = 'changelog-json'
26.         ...
27. );

复制代码

使用 Regular Join 进行多路 Join，数据表打宽操作如下所示

1. SELECT o.order_id, i.item_id, d.item_name, d.item_price, o.order_time
2. FROM orders o
3. LEFT JOIN order_item i ON o.order_id = i.order_id
4. LEFT JOIN item_detail d ON i.item_id = d.item_id

复制代码

最终生成的 DAG 图如下所示:

可以发现:
第一个 join (后面统一简称为ijoin1)的条件是 order_id，该 join 的两个输入会以 order_id 进行hash，具有相同 order_id 的数据能够被发送到同一个 subtask
第二个 join (后面统一简称为 join2)的条件则是 item_id， 该 join 的两个输入会以 item_id 进行hash，具有相同 item_id 的数据则会被发送到同一个 subtask.
正常情况下， 具有相同 order_id 的数据， 一定具有相同的 item_id，但由于上面的示例代码中，我们使用的是 left join 的写法， 即使没有 join 上， 也会输出为 null 的数据，这样可能导致了最终结果的不确定性。
以下面的数据为示例，再详细说明一下：
TABLE  orders
order_idorder_timeid_0012022-06-03 00:00:00TABLE order_item
order_iditem_idid_001item_001TABLE item_detail
item_iditem_nameitem_priceitem_001类目110输出数据如下：
1) 表示输出数据的并发
+I 表示数据的属性 (+I,  , -D, -U, +U)
第一个 JOIN 输出

1. 1) +I(id_001, null, 2022-06-03 00:00:00)
2. 1) -D(id_001, null, 2022-06-03 00：00：00)
3. 1) +I(id_001, item_001, 2022-06-03 00:00:00)

复制代码

第二个 JOIN 输出

1. 1) +I(id_001, null, null, null, 2022-06-03 00:00:00)
2. 1) -D(id_001, null, null, null, 2022-06-03 00:00:00)
3. 2) +I(id_001, item_001, null, null, 2022-06-03 00:00:00)
4. 2) -D(id_001, item_001, null, null, 2022-06-03 00:00:00)
5. 2) +I(id_001, item_001, 类目1, 10, 2022-06-03 00:00:00)

复制代码

以上结果只是上述作业可能出现的情况之一，实际运行时并不一定会出现。 我们可以发现 join1 结果发送到 join2 之后， 相同的 order_id 并不一定会发送到同一个 subtask，因此当数据经过了 join2， 相同的 order_id 的数据会落到不同的并发，这样在后续的数据处理中， 有非常大的概率会导致最终结果的不确定性。
我们再细分以下场景考虑， 假设经过 join2 之后的结果为 join_view：

• 假设 join2 之后，我们基于 item_id  进行聚合， 统计相同类目的订单数
  

1. SELECT item_id, sum(order_id)
2. FROM join_view
3. GROUP BY item_id

复制代码

很显然， 上述的乱序问题并不会影响这段逻辑的结果， item_id 为 null 的数据会进行计算， 但并不会影响 item_id 为 item_001 的结果.

• 假设 join2 之后， 我们将结果直接写入 MySQL,  MySQL 主键为 order_id
  

1. CREATE TABLE MySQL_Sink (
2.         order_id VARCHAR,
3.         item_id VARCHAR,
4.         item_name VARCHAR,
5.         item_price INT,
6.         order_time TIMESTAMP,
7.         PRIMARY KEY (order_id) NOT ENFORCED
8. ) with (
9.         'connector' = 'jdbc'
10. );
12. INSERT INTO MySQL_Sink SELECT * FROM JOIN_VIEW;

复制代码

由于我们在 Sink connector 中未单独设置并发， 因此 sink 的并发度是和 join2 的并发是一样的， 因此 join2 的输出会直接发送给 sink 算子， 并写入到 MySQL 中。
由于是不同并发同时在写 MySQL ，所以实际写 MySQL的顺序可能如下所示：

1. 2) +I(id_001, item_001, null, null, 2022-06-03 00:00:00)
2. 2) -D(id_001, item_001, null, null, 2022-06-03 00:00:00)
3. 2) +I(id_001, item_001, 类目1, 10, 2022-06-03 00:00:00)
4. 1) +I(id_001, null, null, 2022-06-03 00:00:00)
5. 1) -D(id_001, null, null, 2022-06-03 00:00:00)

复制代码

很显然， 最终结果会是 为 空， 最终写入的是一条 delete 数据

• 假设 join2 之后， 我们将结果直接写入 MySQL,  主键为 order_id, item_id, item_name
  

1. CREATE TABLE MySQL_Sink (
2.         order_id VARCHAR,
3.         item_id VARCHAR,
4.         item_name VARCHAR,
5.         item_price INT,
6.         order_time TIMESTAMP,
7.         PRIMARY KEY (order_id) NOT ENFORCED
8. ) with (
9.         'connector' = 'jdbc'
10. );
12. INSERT INTO MySQL_Sink SELECT * FROM JOIN_VIEW;

复制代码

和示例2一样， 我们未单独设置 sink 的并发， 因此数据会之间发送到 sink 算子， 假设写入  MySQL 的顺序和示例2一样：

1. 2) +I(id_001, item_001, null, null, 2022-06-03 00:00:00)
2. 2) -D(id_001, item_001, null, null, 2022-06-03 00:00:00)
3. 2) +I(id_001, item_001, 类目1, 10, 2022-06-03 00:00:00)
4. 1) +I(id_001, null, null, null, 2022-06-03 00:00:00)
5. 1) -D(id_001, null, null, null, 2022-06-03 00:00:00)

复制代码

最终结果会是

1. 2) +I(id_001, item_001, 类目1, 2022-06-03 00:00:00)

复制代码

由于 MySQL 的主键是 order_id, item_id, item_name 所以最后的 -D 记录并不会删除 subtask 2 写入的数据， 这样最终的结果是正确的。

• 假设 join2 之后， 我们将结果写入 kafka，写入格式为 changelog-json , 下游作业消费 kafka 并进行处理
  

1. CREATE TABLE kafka_sink (
2.         order_id VARCHAR,
3.         item_id VARCHAR,
4.         item_name VARCHAR,
5.         item_price INT,
6.         order_time TIMESTAMP,
7.         PRIMARY KEY (order_id, item_id, item_name) NOT ENFORCED
8. ) with (
9.         'connector' = 'kafka',
10.         'format' = 'changelog-json',
11.         'topic' = 'join_result'
12. );
14. INSERT INTO kafka_sink select * from JOIN_VIEW;

复制代码

默认，如果不设置 partitioner， kafka sink 会以我们在 DDL 中配置的主键生成对应的 hash key， 用于通过 hash 值生成 partition id。
有一点我们需要注意， 由于 join2 的输出已经在不同的并发了， 所以无论 kafka_sink  选择以  order_id 作为唯一的主键， 还是以  order_id, item_id, item_name 作为主键， 我们都无法控制不同并发写入 kafka 的顺序， 我们只能确保相同的并发的数据能够有序的被写入 kafka 的同一 partition 。

• 如果设置 order_id 为主键， 我们可以保证上述的所有数据能够被写入同一个 partition
  
• 如果设置 order_id, item_id, item_name 则上面不同并发的输出可能会被写入到不同的 partition
  

所以，我们需要关注的是， 当数据写入 kafka 之后， 下游怎么去处理这一份数据:

• 基于 order_id 进行去重，并按天聚合，计算当天的累加值。
  以下面的 SQL 为例， 下游在消费 kafka 时， 为了避免数据重复， 先基于 order_id 做了一次去重， 用 order_id  作为分区条件,  基于proctime() 进行去重 (增加table.exec.source.cdc-events-duplicate 该参数， 框架会自动生成去重算子).
  

1. -- https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/config/
2. set 'table.exec.source.cdc-events-duplicate'='true';
4. CREATE TABLE kafka_source (
5.         order_id VARCHAR,
6.         item_id VARCHAR,
7.         item_name VARCHAR,
8.         item_price INT,
9.         order_time TIMESTAMP,
10.         PRIMARY KEY (order_id) NOT ENFORCED
11. ) with (
12.         'connector' = 'kafka',
13.         'format' = 'changelog-json',
14.         'topic' = 'join_result'
15. );
17. -- 按order_time 聚合， 计算每天的营收
19. SELECT DATE_FORMAT(order_time, 'yyyy-MM-dd'), sum(item_price)
20. FROM kafka_source
21. GROUP BY DATE_FORMAT(order_time, 'yyyy-MM-dd')

复制代码

结果上述的计算，我们预计结果会如何输出：
去重之后可能的输出为:

1. 1) +I(id_001, item_001, null, null, 2022-06-03 00:00:00)
2. 1) -D(id_001, item_001, null, null, 2022-06-03 00:00:00)
3. 1) +I(id_001, item_001, 类目1, 10, 2022-06-03 00:00:00)
4. 1) -D(id_001, item_001, 类目1, 10, 2022-06-03 00:00:00)
5. 1) +I(id_001, item_001, null, null, 2022-06-03 00:00:00)
6. 1) -D(id_001, item_001, null, null, 2022-06-03 00:00:00)

复制代码

经过聚合算子算子:

1. 1) +I(2022-06-03, null)
2. 1) -D(2022-06-03, null)
3. 1) +I(2022-06-03, 10)
4. 1) -D(2022-06-03, 10)
5. 1) +I(2022-06-03, 0)
6. 1) -D(2022-06-03, 0)
7. 1) +I(2022-06-03, null)
8. 1) -D(2022-06-03, null)

复制代码

可以发现，最终输出结果为2022-06-03, null，本文列举的示例不够完善， 正常情况下， 当天肯定会有其他的记录， 结果当天的结果可能不会为 null， 但我们可以知道的是，由于数据的乱序， 数据和实际结果已经不准确了。
2) 基于order_id, item_id, item_name 去重，之后按天聚合，计算当天的累加值。

1. -- https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/config/
2. set 'table.exec.source.cdc-events-duplicate'='true';
4. CREATE TABLE kafka_source (
5.         order_id VARCHAR,
6.         item_id VARCHAR,
7.         item_name VARCHAR,
8.         item_price INT,
9.         order_time TIMESTAMP,
10.         PRIMARY KEY (order_id, item_id, item_name) NOT ENFORCED
11. ) with (
12.         'connector' = 'kafka',
13.         'format' = 'changelog-json',
14.         'topic' = 'join_result'
15. );
17. -- 按order_time 聚合， 计算每天的营收
19. SELECT DATE_FORMAT(order_time, 'yyyy-MM-dd'), sum(item_price)
20. FROM kafka_source
21. GROUP BY DATE_FORMAT(order_time, 'yyyy-MM-dd')

复制代码

去重之后的输出:

1. 1) +I(id_001, item_001, null, null, 2022-06-03 00:00:00)
2. 1) -D(id_001, item_001, null, null, 2022-06-03 00:00:00)
3. 1) +I(id_001, item_001, 类目1, 10, 2022-06-03 00:00:00)
4. 2) +I(id_001, item_001, null, null, 2022-06-03 00:00:00)
5. 2) -D(id_001, item_001, null, null, 2022-06-03 00:00:00)

复制代码

由于我们主键设置的是 order_id, item_id, item_name 所以
(id_001, item_001, null, null, 2022-06-03 00:00:00) 和
1) +I(id_001, item_001, 类目1, 10, 2022-06-03 00:00:00) 是不同的主键， 所以并不会互相影响。
经过聚合之后的结果：

1. 1) +I(2022-06-03, null)
2. 1) -D(2022-06-03, null)
3. 1) +I(2022-06-03, 10)
4. 1) -D(2022-06-03, 10)
5. 1) +I(2022-06-03, 10)
6. 1) -D(2022-06-03, 10)
7. 1) +I(2022-06-03, 10)

复制代码

以上就是最终结果的输出， 可以发现我们最终的结果是没有问题的。
原因分析

下图是原始作业的数据流转变化情况
graph LRorders(orders) --> |hash:order_id| join1(join1)order_item(order_item) -->|hash:order_id| join1 join1 --> |hash:item_id| join2(join2)item_detail(item_detail) --> |hash:item_id| join2

• A 基于 item_id 聚合， 计算相同类目的订单数 (结果正确)
  

graph LRjoin2(join2) --> |hash:item_id| group(group count:order_id)

• B 将join的数据 sink 至 MySQL (主键为 order_id) (结果错误)
  

graph LRjoin2(join2) --> |forward| sink(sink)sink --> |jdbc send| MySQL(MySQL primarykey:order_id)

• C 将join的数据 sink 至 MySQL (主键为 order_id, item_id, item_name) (结果正确)
  

graph LRjoin2(join2) --> |forward| sink(sink)sink --> |jdbc send| MySQL(MySQL primarykey:order_id)

• D 将 join 的数据 sink 至 kafka， 下游消费 kafka 数据并进行去重处理， 下游处理时，又可以分为两种情况。
  
  
  
  • D-1 按 order_id 分区并去重 (结果错误)
    
  • D-2 按 order_id, item_id, item_name 分区并去重 (结果正确)
    
  
  

graph TDjoin2(join2) --> |forward| sink(sink)sink --> |kafka client| kafka(Kafka fixed partitioner)kafka --> |hash:order_id| rank(rank orderby:proctime)rank --> |"hash:date_format(order_time, 'yyyy-MM-dd')"| group("group agg:sum(item_price)")graph TDjoin2(join2) --> |forward| sink(sink key)sink --> |kafka client| kafka(Kafka fixed partitioner)kafka --> |hash:order_id+item_id+item_name| rank(rank orderby:proctime)rank --> |"hash:date_format(order_time, 'yyyy-MM-dd')"| group("group agg:sum(item_price)")从上面 A, B, C, D-1, D-2 这四种 case， 我们不难发现， 什么情况下会导致错误的结果， 什么情况下不会导致错误的结果， 关键还是要看每个 task 之间的 hash 规则。
case B 产生乱序主要原因时在 sink operator， hash的条件由原来的 order_id+item_id_item_name 变成了 order_id
case D-1 产生乱序主要发生在去重的  operator， hash 的规则由原来的 order_id+item_id+item_name 变为了 order_id
我们大概能总结以下几点经验

• Flink 框架在可以保证 operator 和 operator hash 时， 一定是可以保证具有相同 hash 值的数据的在两个 operator 之间传输顺序性
  
• Flink 框架无法保证数据连续多个 operator hash 的顺序， 当 operator 和 operator 之间的 hash 条件发生变化， 则有可能出现数据的顺序性问题。
  
• 当 hash 条件由少变多时， 不会产生顺序问题， 当 hash 条件由多变少时， 则可能会产生顺序问题。
  

总结

大多数业务都是拿着原来的实时任务， 核心逻辑不变，只是把原来的 Hive 替换成 消息队列的 Source 表， 这样跑出来的结果，一般情况下就很难和离线对上，虽然流批一体是 Flink 的优势， 但对于某些 case ， 实时的结果和离线的结果还是会产生差异， 因此我们在编写 FlinkSQL 代码时， 一定要确保数据的准备性， 在编写代码时，一定要知道我们的数据大概会产生怎样的流动， 产生怎样的结果， 这样写出来的逻辑才是符合预期的。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！