聊聊流式数据湖Paimon(四)

这篇具有很好参考价值的文章主要介绍了聊聊流式数据湖Paimon(四)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Partial Update

数据打宽

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

--FlinkSQL参数设置
set
    `table.dynamic-table-options.enabled` = `true`;

SET
    `env.state.backend` = `rocksdb`;

SET
    `execution.checkpointing.interval` = `60000`;

SET
    `execution.checkpointing.tolerable-failed-checkpoints` = `3`;

SET
    `execution.checkpointing.min-pause` = `60000`;

--创建Paimon catalog
CREATE CATALOG paimon WITH (
    'type' = 'paimon',
    'metastore' = 'hive',
    'uri' = 'thrift://localhost:9083',
    'warehouse' = 'hdfs://paimon',
    'table.type' = 'EXTERNAL'
);

--创建Partial update结果表
CREATE TABLE if not EXISTS paimon.dw.order_detail (
    `order_id` string,
    `product_type` string,
    `plat_name` string,
    `ref_id` bigint,
    `start_city_name` string,
    `end_city_name` string,
    `create_time` timestamp(3),
    `update_time` timestamp(3),
    `dispatch_time` timestamp(3),
    `decision_time` timestamp(3),
    `finish_time` timestamp(3),
    `order_status` int,
    `binlog_time` bigint,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'bucket' = '20',
    -- 指定20个bucket
    'bucket-key' = 'order_id',
    -- 记录排序字段
    'sequence.field' = 'binlog_time',
    -- 选择 full-compaction ,在compaction后产生完整的changelog
    'changelog-producer' = 'full-compaction',
    -- compaction 间隔时间
    'changelog-producer.compaction-interval' = '2 min',
    'merge-engine' = 'partial-update',
    -- 忽略DELETE数据,避免运行报错
    'partial-update.ignore-delete' = 'true'
);

INSERT INTO
    paimon.dw.order_detail 
-- order_info表提供主要字段
SELECT
    order_id,
    product_type,
    plat_name,
    ref_id,
    cast(null as string) as start_city_name,
    cast(null as string) as end_city_name,
    create_time,
    update_time,
    dispatch_time,
    decision_time,
    finish_time,
    order_status,
    binlog_time
FROM
    paimon.ods.order_info
    /*+ OPTIONS ('scan.mode'='latest') */
union
all 
-- order_address表提供城市字段
SELECT
    order_id,
    cast(null as string) as product_type,
    cast(null as string) as plat_name,
    cast(null as bigint) as ref_id,
    start_city_name,
    end_city_name,
    cast(null as timestamp(3)) as create_time,
    cast(null as timestamp(3)) as update_time,
    cast(null as timestamp(3)) as dispatch_time,
    cast(null as timestamp(3)) as decision_time,
    cast(null as timestamp(3)) as finish_time,
    cast(null as int) as order_status,
    binlog_time
FROM
    paimon.ods.order_address
    /*+ OPTIONS ('scan.mode'='latest') */
;

完整的Changlog

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

Sequence-Group

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

public void testSequenceGroup() {
    sql(
            "CREATE TABLE SG ("
                    + "k INT, a INT, b INT, g_1 INT, c INT, d INT, g_2 INT, PRIMARY KEY (k) NOT ENFORCED)"
                    + " WITH ("
                    + "'merge-engine'='partial-update', "
                    + "'fields.g_1.sequence-group'='a,b', "
                    + "'fields.g_2.sequence-group'='c,d');");

    sql("INSERT INTO SG VALUES (1, 1, 1, 1, 1, 1, 1)");

    // g_2 should not be updated
    sql("INSERT INTO SG VALUES (1, 2, 2, 2, 2, 2, CAST(NULL AS INT))");

    // select *
    assertThat(sql("SELECT * FROM SG")).containsExactlyInAnyOrder(Row.of(1, 2, 2, 2, 1, 1, 1));

    // projection
    assertThat(sql("SELECT c, d FROM SG")).containsExactlyInAnyOrder(Row.of(1, 1));

    // g_1 should not be updated
    sql("INSERT INTO SG VALUES (1, 3, 3, 1, 3, 3, 3)");

    assertThat(sql("SELECT * FROM SG")).containsExactlyInAnyOrder(Row.of(1, 2, 2, 2, 3, 3, 3));

    // d should be updated by null
    sql("INSERT INTO SG VALUES (1, 3, 3, 3, 2, 2, CAST(NULL AS INT))");
    sql("INSERT INTO SG VALUES (1, 4, 4, 4, 2, 2, CAST(NULL AS INT))");
    sql("INSERT INTO SG VALUES (1, 5, 5, 3, 5, CAST(NULL AS INT), 4)");

    assertThat(sql("SELECT a, b FROM SG")).containsExactlyInAnyOrder(Row.of(4, 4));
    assertThat(sql("SELECT c, d FROM SG")).containsExactlyInAnyOrder(Row.of(5, null));
}

其作用是:

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

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 进行计算,明确了数据版本,从而实现计算的一致性。
聊聊流式数据湖Paimon(四)

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

数据对齐

将 Checkpoint 插入到两个 Snapshot 的数据之间。如果当前的 Snapshot 还没有完全被消费,这个 Checkpoint 的触发会被推迟,从而实现按照 Snapshot 对数据进行划分和对齐。
聊聊流式数据湖Paimon(四)
实现分为两个部分。

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

数据血缘

概念

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

实现

在checkpoint的提交时将数据的血缘关系写入到System Table,记录血缘关系。
聊聊流式数据湖Paimon(四)文章来源地址https://www.toymoban.com/news/detail-776371.html

到了这里,关于聊聊流式数据湖Paimon(四)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 基于 Flink SQL 和 Paimon 构建流式湖仓新方案

    目录 1. 数据分析架构演进 2. Apache Paimon 3. Flink + Paimon 流式湖仓 Consumer 机制 Changelog 生成​编辑

    2024年02月04日
    浏览(34)
  • 【大数据】-- 创建 Paimon 外部表

           如今,在数据湖三剑客(delta lake、hudi、iceberg)之上,又新出一派: apache paimon。我们恰好在工作中遇到,以下介绍在 dataworks 上,使用 maxcompute odps sql 创建 apache paimon 外部表的一些操作和注意事项。参考:创建MaxCompute Paimon外部表_云原生大数据计算服务 MaxCompute(Max

    2024年03月13日
    浏览(37)
  • Flink + Paimon数据 CDC 入湖最佳实践

    Apache Paimon 最典型的场景是解决了 CDC (Change Data Capture)数据的入湖,看完这篇文章可以了解到: 1、为什么 CDC 入Hive迁移到 Paimon? 2、CDC 入 Paimon 怎么样做到成本最低? 3、Paimon 对比 Hudi有什么样的优势?  Paimon 从 CDC 入湖场景出发,希望提供给你 简单、低成本、低延时 的

    2024年01月16日
    浏览(44)
  • Paimon+StarRocks 湖仓一体数据分析方案

    摘要:本文整理自阿里云高级开发工程师曾庆栋(曦乐)在 Streaming Lakehouse Meetup 的分享。内容主要分为四个部分: 传统数据仓库分析实现方案简介 Paimon+StarRocks 构建湖仓一体数据分析实现方案 StarRocks 与 Paimon 结合的使用方式与实现原理 StarRocks 社区湖仓分析未来规划 点击查

    2024年02月10日
    浏览(42)
  • doris - 数仓 拉链表 按天全量打宽表性能优化

    1、 业务历史数据可以变更 2、拉链表按天打宽 3、拉链表模型分区字段设计不合理,通用的过滤字段没有作为分区分桶字段 4、拉链表表数据量略大、模型数据分区不合理和服务器资源限制,计算任务执行超时【3-4年,用户数:132W】 5、基于拉链表打宽后的天表行转列【 最多

    2024年02月14日
    浏览(50)
  • 4 Paimon数据湖之Hive Catalog的使用

    更多Paimon数据湖内容请关注 :https://edu.51cto.com/course/35051.html Paimon提供了两种类型的Catalog: Filesystem Catalog 和 Hive Catalog 。 Filesystem Catalog:会把元数据信息存储到文件系统里面。 Hive Catalog:则会把元数据信息存储到Hive的Metastore里面,这样就可以直接在Hive中访问Paimon表了。注

    2024年02月02日
    浏览(43)
  • 流数据湖平台Apache Paimon(四)集成 Hive 引擎

    前面与Flink集成时,通过使用 paimon Hive Catalog,可以从 Flink 创建、删除、查询和插入到 paimon 表中。这些操作直接影响相应的Hive元存储。以这种方式创建的表也可以直接从 Hive 访问。 更进一步的与 Hive 集成,可以使用 Hive SQL创建、查询Paimon表。 Paimon 目前支持 Hive 3.1、2.3、2

    2024年02月14日
    浏览(55)
  • 流数据湖平台Apache Paimon(三)Flink进阶使用

    2.9.1 写入性能 Paimon的写入性能与检查点密切相关,因此需要更大的写入吞吐量: 增加检查点间隔,或者仅使用批处理模式。 增加写入缓冲区大小。 启用写缓冲区溢出。 如果您使用固定存储桶模式,请重新调整存储桶数量。 2.9.1.1 并行度 建议sink的并行度小于等于bucket的数量

    2024年02月09日
    浏览(32)
  • 流数据湖平台Apache Paimon(二)集成 Flink 引擎

    Paimon目前支持Flink 1.17, 1.16, 1.15 和 1.14。本课程使用Flink 1.17.0。 环境准备 2.1.1 安装 Flink 1)上传并解压Flink安装包 tar -zxvf flink-1.17.0-bin-scala_2.12.tgz -C /opt/module/ 2)配置环境变量 2.1.2 上传 jar 包 1)下载并上传Paimon的jar包 jar包下载地址:https://repository.apache.org/snapshots/org/apache/pa

    2024年02月09日
    浏览(44)
  • 【Flink SQL API体验数据湖格式之paimon】

    随着大数据技术的普及,数据仓库的部署方式也在发生着改变,之前在部署数据仓库项目时,首先想到的是选择国外哪家公司的产品,比如:数据存储会从Oracle、SqlServer中或者Mysql中选择,ETL工具会从Informatica、DataStage或者Kettle中选择,BI报表工具会从IBM cognos、Sap Bo或者帆软中

    2024年02月04日
    浏览(38)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包