0
点赞
收藏
分享

微信扫一扫

C++ —— 内存管理

东言肆语 03-22 09:00 阅读 3

        乱序数据处理对于实时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 的数据。除此之外,我们还要讲该数据记录在状态中,以方便后续对侧数据来做镜像的关联处理。

举报

相关推荐

【c++】C/C++内存管理

【C++】C/C++内存管理

c++内存管理

C/C++内存管理

【C++】内存管理

C++内存管理

[C/C++]——内存管理

C/C++内存管理详解

0 条评论