乱序数据处理对于实时ETL至关重要，处理不好将会导致数据不一致场景发生。对于数据乱序场景，一般工程师已知上游数据乱序会对本身消费数据产生影响，但不一定晓得的是，一个SQL本身也可能造成数据乱序，严格意义上的数据乱序是无法避免的。本文讨论的是在SQL开发过程中，由于考虑不当导致数据乱序的场景。
先来了解下示例数据，我们以交易场景为例，这里有三张表，分别是提交订单表、提交订单明细表和商品维表，如下：
-- 提单表：一个订单一行
create table ods_submit_order(order_id bigint comment '订单ID',create_time string comment '创建时间',primary key(order_id) not enforced
);
-- 提单明细表：一个订单一个商品一行
create table ods_submit_order_product(order_id bigint comment '订单ID',product_id bigint comment '商品ID',submit_amt double comment '订单金额',create_time string comment '创建时间',primary key(order_id,product_id) not enforced
);
-- 商品维表：一个商品一行
create table dim_product(product_id bigint comment '商品ID',product_name string comment '商品名称',create_time string comment '创建时间',primary key (product_id) not enforced
);本文需要讨论的造成乱序的SQL如下：
insert into fact_ord_submit_order_dd
select t1.order_id as order_id,t2.product_id as product_id,t2.submit_amt as submit_amt,t3.product_name as product_name
from ods_submit_order t1
left join ods_submit_order_product t2 on t1.order_id = t2.order_id
left join dim_product t3 on t2.product_id = t3.product_id
;测试SQL对应的DAG图如下：
这里假设source数据是保序的，source并发为1，对应一个subtask，join并发2，对应两个subtask处理join逻辑，根据DAG图分析：
-
Join-one：首先t1 left join t2通过order_id进行左外连接，这时Flink内部会根据order_id的hash值将相同order_id的数据shuffer到相同的subtask上执行。 -
Join-two:其次t2 left join t3通过product_id进行左外连接，这时Flink内部会根据product_id的hash值将相同product_id的数据shuffer到相同的subtask上执行。
可以发现，这个SQL进行了两次shuffer过程，而且两次shuffer的key都不一样。
Join-one可能对应的数据流如下：order_id=1数据都会shuffer到同一个subtask上
p1: +I(1, null, null)
p1: -D(1, null, null)
p1: +I(1, 1, 10.0) Join-two可能对应的数据流如下：product_id=1和为null的数据被shuffer到了不同的subtask上进行处理
p1: +I(1, null, null, null)
p1: -D(1, null, null, null)
p2: +I(1, 1, 10.0, null)
p2: -D(1, 1, 10.0, null)
p2: +I(1, 1, 10.0, java) 从Join-two的输出结果可以发现，当前输出由两个subtask任务处理，数据最终结果为+I(1, 1, 10.0, java),所以这时候数据发往下游表fact_ord_submit_order_dd，由于不同并发task处理能力不一样，发送数据顺序可能不一样，将会导致结果也不一样。分两种情况来看(这里left join会产生变更流)，如下：
-
场景一：结果表fact_ord_submit_order_dd为
order_id：-
场景1：比如p1先发送下游、p2后发送下游；这时候如果下游设置order_id为主键，最终结果为
+I(1, 1, 10.0, java)，下游数据正确。即下游消费顺序：p1: +I(1, null, null, null) p1: -D(1, null, null, null) p2: +I(1, 1, 10.0, null) p2: -D(1, 1, 10.0, null) p2: +I(1, 1, 10.0, java)
-
场景2：如果p2先发送下游，然后p1发送下游；由于接收顺序与
场景1不一致，最终接收结果为-D(1, null, null, null)，数据出现异常，归根到底是由于left join导致数据乱序原因。即下游消费顺序：p2: +I(1, 1, 10.0, null) p2: -D(1, 1, 10.0, null) p2: +I(1, 1, 10.0, java) p1: +I(1, null, null, null) p1: -D(1, null, null, null)
-
-
场景二：结果表fact_ord_submit_order_dd为
order_id,product_id：-
场景1：比如p1先发送下游、p2后发送下游；这时候如果下游设置order_id和product_id为主键，最终结果为
+I(1, 1, 10.0, java)，下游数据正确。即下游消费顺序：p1: +I(1, null, null, null) p1: -D(1, null, null, null) p2: +I(1, 1, 10.0, null) p2: -D(1, 1, 10.0, null) p2: +I(1, 1, 10.0, java) -
场景2：如果p2先发送下游，然后p1发送下游；由于接收顺序与
场景1不一致，但下游设置order_id和product_id为主键，最终结果为+I(1, 1, 10.0, java)任然不受-D(1, null, null, null)的影响，下游数据正确。即下游消费顺序：p2: +I(1, 1, 10.0, null) p2: -D(1, 1, 10.0, null) p2: +I(1, 1, 10.0, java) p1: +I(1, null, null, null) p1: -D(1, null, null, null)
-
这个例子从两次left join场景出发，我们可以发现，下游sink表主键设置会影响最终的数据准确性。当然，业务上结果表的主键也应该为order_id和product_id作为联合主键，这里order_id假设为主键对于场景ETL来说并不合适，这里仅仅是为了说明问题。
下面让我们尝试总结一下这个Regular Join场景的执行逻辑：在流式处理数据的过程中，当本侧到来一条新的数据时，我们无法预测对侧是否在之后还会到来能够和该数据关联上的数据，且考虑到时效性，我们也无法一直等待右侧所有数据到齐后再关联下发，因此 Flink 的处理方式是先将当前数据和对侧已经到来过的所有数据（如果设置了 TTL，则是对应 TTL 时间段的数据）进行关联计算，并将关联结果下发，如果是 Outer Join，则还要考虑关联不上需要下发一条对侧为 null 的数据。除此之外，我们还要讲该数据记录在状态中，以方便后续对侧数据来做镜像的关联处理。
