聊聊流式数据湖Paimon(四)

Partial Update

数据打宽

通过不同的流写不同的字段，打宽了数据的维度，填充了数据内容；如下所示：

1. --FlinkSQL参数设置
2. set
3.     `table.dynamic-table-options.enabled` = `true`;
5. SET
6.     `env.state.backend` = `rocksdb`;
8. SET
9.     `execution.checkpointing.interval` = `60000`;
11. SET
12.     `execution.checkpointing.tolerable-failed-checkpoints` = `3`;
14. SET
15.     `execution.checkpointing.min-pause` = `60000`;
17. --创建Paimon catalog
18. CREATE CATALOG paimon WITH (
19.     'type' = 'paimon',
20.     'metastore' = 'hive',
21.     'uri' = 'thrift://localhost:9083',
22.     'warehouse' = 'hdfs://paimon',
23.     'table.type' = 'EXTERNAL'
24. );
26. --创建Partial update结果表
27. CREATE TABLE if not EXISTS paimon.dw.order_detail (
28.     `order_id` string,
29.     `product_type` string,
30.     `plat_name` string,
31.     `ref_id` bigint,
32.     `start_city_name` string,
33.     `end_city_name` string,
34.     `create_time` timestamp(3),
35.     `update_time` timestamp(3),
36.     `dispatch_time` timestamp(3),
37.     `decision_time` timestamp(3),
38.     `finish_time` timestamp(3),
39.     `order_status` int,
40.     `binlog_time` bigint,
41.     PRIMARY KEY (order_id) NOT ENFORCED
42. ) WITH (
43.     'bucket' = '20',
44.     -- 指定20个bucket
45.     'bucket-key' = 'order_id',
46.     -- 记录排序字段
47.     'sequence.field' = 'binlog_time',
48.     -- 选择 full-compaction ，在compaction后产生完整的changelog
49.     'changelog-producer' = 'full-compaction',
50.     -- compaction 间隔时间
51.     'changelog-producer.compaction-interval' = '2 min',
52.     'merge-engine' = 'partial-update',
53.     -- 忽略DELETE数据，避免运行报错
54.     'partial-update.ignore-delete' = 'true'
55. );
57. INSERT INTO
58.     paimon.dw.order_detail
59. -- order_info表提供主要字段
60. SELECT
61.     order_id,
62.     product_type,
63.     plat_name,
64.     ref_id,
65.     cast(null as string) as start_city_name,
66.     cast(null as string) as end_city_name,
67.     create_time,
68.     update_time,
69.     dispatch_time,
70.     decision_time,
71.     finish_time,
72.     order_status,
73.     binlog_time
74. FROM
75.     paimon.ods.order_info
76.     /*+ OPTIONS ('scan.mode'='latest') */
77. union
78. all
79. -- order_address表提供城市字段
80. SELECT
81.     order_id,
82.     cast(null as string) as product_type,
83.     cast(null as string) as plat_name,
84.     cast(null as bigint) as ref_id,
85.     start_city_name,
86.     end_city_name,
87.     cast(null as timestamp(3)) as create_time,
88.     cast(null as timestamp(3)) as update_time,
89.     cast(null as timestamp(3)) as dispatch_time,
90.     cast(null as timestamp(3)) as decision_time,
91.     cast(null as timestamp(3)) as finish_time,
92.     cast(null as int) as order_status,
93.     binlog_time
94. FROM
95.     paimon.ods.order_address
96.     /*+ OPTIONS ('scan.mode'='latest') */
97. ;

复制代码

完整的Changlog

Paimon中的表被多流填充数据且打宽维度后，支持流读、批读的方式提供完整的Changelog给下游。
Sequence-Group

配置：'fields.G.sequence-group'='A,B'
由字段G控制是否更新字段A, B；总得来说，G的值如果为null或比更新值大将不更新A,B；如下单测

1. public void testSequenceGroup() {
2.     sql(
3.             "CREATE TABLE SG ("
4.                     + "k INT, a INT, b INT, g_1 INT, c INT, d INT, g_2 INT, PRIMARY KEY (k) NOT ENFORCED)"
5.                     + " WITH ("
6.                     + "'merge-engine'='partial-update', "
7.                     + "'fields.g_1.sequence-group'='a,b', "
8.                     + "'fields.g_2.sequence-group'='c,d');");
10.     sql("INSERT INTO SG VALUES (1, 1, 1, 1, 1, 1, 1)");
12.     // g_2 should not be updated
13.     sql("INSERT INTO SG VALUES (1, 2, 2, 2, 2, 2, CAST(NULL AS INT))");
15.     // select *
16.     assertThat(sql("SELECT * FROM SG")).containsExactlyInAnyOrder(Row.of(1, 2, 2, 2, 1, 1, 1));
18.     // projection
19.     assertThat(sql("SELECT c, d FROM SG")).containsExactlyInAnyOrder(Row.of(1, 1));
21.     // g_1 should not be updated
22.     sql("INSERT INTO SG VALUES (1, 3, 3, 1, 3, 3, 3)");
24.     assertThat(sql("SELECT * FROM SG")).containsExactlyInAnyOrder(Row.of(1, 2, 2, 2, 3, 3, 3));
26.     // d should be updated by null
27.     sql("INSERT INTO SG VALUES (1, 3, 3, 3, 2, 2, CAST(NULL AS INT))");
28.     sql("INSERT INTO SG VALUES (1, 4, 4, 4, 2, 2, CAST(NULL AS INT))");
29.     sql("INSERT INTO SG VALUES (1, 5, 5, 3, 5, CAST(NULL AS INT), 4)");
31.     assertThat(sql("SELECT a, b FROM SG")).containsExactlyInAnyOrder(Row.of(4, 4));
32.     assertThat(sql("SELECT c, d FROM SG")).containsExactlyInAnyOrder(Row.of(5, null));
33. }

复制代码

其作用是：

• 在多个数据流更新期间的无序问题。每个数据流都定义自己的序列组。
  
• 真正的部分更新，而不仅仅是非空值的更新。
  
• 接受删除记录来撤销部分列。
  

Changelog-Producer

Paimon通过Changelog-Producer支持生成changelog，并支持下游以流读、批读的形式读取changelog。
Changelog的生成有多种方式，input、lookup、full-compaction；其生成代价是由低到高。
None

不查找旧值，不额外写Changelog；但会下游任务中通过ChangelogNormalize算子补足Changelog。
Input

不查找旧值，额外写Changelog；适用与CDC的数据源。
Lookup

查找旧值，额外写Changelog；如果不是CDC数据源，需要通过LookupCompaction查找旧值，即在 compaction 的过程中, 会去向高层查找本次新增 key 的旧值, 如果没有查找到, 那么本次的就是新增 key, 如果有查找到, 那么就生成完整的 UB 和 UA 消息。
Full-Compaction

查找旧值，额外写Changelog；在 full compact 的过程中, 其实数据都会被写到最高层, 所以所有 value 的变化都是可以推演出来的.
数据一致性

数据版本

通过Flink的checkpoint机制，生成Snapshot并标记版本，即，一个Snapshot对应数据的一个版本。
比如 Job-A 基于 Table-A 的 Snapshot-20 产出了 Table-B 的 Snapshot-11。Job-B 基于 Table-A 的Snapshot-20产出了 Table-C 的 Snapshot-15。那么 Job-C 的查询就应该基于 Table-B 的 Snapshot-11 和 Table-C 的 Snapshot-15 进行计算，明确了数据版本，从而实现计算的一致性。

生成的snapshot-xx，就是数据的版本号。

数据对齐

将 Checkpoint 插入到两个 Snapshot 的数据之间。如果当前的 Snapshot 还没有完全被消费，这个 Checkpoint 的触发会被推迟，从而实现按照 Snapshot 对数据进行划分和对齐。

实现分为两个部分。

• 在提交阶段，需要去血缘关系表中查询上下游表的一致性版本，并且基于查询结果给对应的上游表设置起始的消费位置。
  
• 在运行阶段，按照消费的 Snapshot 来协调 Checkpoint，在 Flink 的 Checkpoint Coordinator 向 Source 发出 Checkpoint 的请求时，会强制要求将 Checkpoint 插入到两个 Snapshot 的数据之间。如果当前的 Snapshot 还没有完全被消费，这个 Checkpoint 的触发会被推迟，从而实现按照 Snapshot 对数据进行划分和处理。
  

数据血缘

概念

数据从产生到消费的整个流转过程中所经历的各种转换、处理和流动的轨迹。数据血缘提供了数据的来源、去向以及中间处理过程的透明度，帮助用户理解数据如何在系统中被处理和移动，以及数据是如何从原始状态转化为最终的可消费形态。
实现

在checkpoint的提交时将数据的血缘关系写入到System Table，记录血缘关系。


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