EMR Serverless Spark:结合实时计算 Flink 基于 Paimon 实现流批一体

本文涉及的产品
EMR Serverless StarRocks,5000CU*H 48000GB*H
简介: 本文演示了使用实时计算 Flink 版和 Serverless Spark 产品快速构建 Paimon 数据湖分析的流程,包括数据入湖 OSS、交互式查询,以及离线Compact。Serverless Spark完全兼容Paimon,通过内置的DLF的元数据实现了和其余云产品如实时计算Flink版的元数据互通,形成了完整的流批一体的解决方案。同时支持灵活的作业运行方式和参数配置,能够满足实时分析、生产调度等多项需求。

简介

EMR Serverless Spark 版是开源大数据平台 E-MapReduce 基于 Spark 提供的一款全托管、一站式的数据计算平台。它为用户提供任务开发、调试、发布、调度和运维等全方位的产品化服务,显著简化了大数据计算的工作流程,使用户能更专注于数据分析和价值提炼。


实时计算 Flink 版是阿里云提供的全托管 Serverless Flink 云服务,具备实时应用的作业开发、数据调试、运行与监控、自动调优、智能诊断等全生命周期能力。


Apache Paimon 是一种流批统一的数据湖存储格式,结合 Flink 及 Spark 构建流批处理的实时湖仓一体架构。Paimon 创新地将湖格式与 LSM 技术结合起来,给数据湖带来了实时流更新以及完整的流处理能力。


本文通过 Flink Datagen 模拟上游数据源,通过实时计算Flink版完成数据的实时摄取,以 Paimon 湖格式存储;然后通过 Serverless Spark 完成数据的分析。同时依赖 Serverless Spark 的调度能力,配置离线的 Compact 作业完成对 Paimon 表的湖表管理。


实时计算 Flink 版实时入湖

  1. 创建 Paimon Catalog,元数据存储类型选择 dlf(通过DLF实现统一的元数据管理,实现多引擎无缝衔接),Catalog 完整配置如下所示。
catalog name: paimon
metastore: dlf
warehouse: oss://test/warehouse
dlf.catalog.id: x
dlf.catalog.accessKeyId: x
dlf.catalog.accessKeySecret: x
dlf.catalog.endpoint: dlf-vpc.cn-hangzhou.aliyuncs.com
dlf.catalog.region: cn-hangzhou


  1. 创建 Paimon 表,按日期和小时分区,其中 category 模拟为 1~10 的随机数,用作后续执行 zorder 的列。设置表属性write-onlytrue,关闭自动Compact,提升写效率。
create table if not exists `paimon`.`test_paimon_db`.`test_append_tbl`
(
    id        string
    ,data     string
    ,category int
    ,ts       string
    ,dt       string
    ,hh       string
) 
partitioned by (dt, hh)
with (
    'write-only' = 'true'
)
;


  1. 启动入湖作业,通过 datagen 源源不断生成数据写入 Paimon 表中,流作业完整 SQL 如下:
CREATE TEMPORARY TABLE datagen
(
    id        string
    ,data     string
    ,category int
)
WITH (
    'connector' = 'datagen'
    ,'rows-per-second' = '100'
    ,'fields.category.kind' = 'random'
    ,'fields.category.min' = '1'
    ,'fields.category.max' = '10'
)
;
INSERT INTO `paimon`.`test_paimon_db`.`test_append_tbl`
SELECT
    id
    ,data
    ,category
    ,cast(LOCALTIMESTAMP as string) as ts
    ,cast(CURRENT_DATE as string) as dt
    ,cast(hour(LOCALTIMESTAMP) as string) as hh
from datagen
;

作业示意图:

Serverless Spark 交互式查询

Serverless Spark 支持交互式查询和作业提交两种方式,前者可用于快速查询调试,后者提供任务开发、调试、调度和运维等全生命周期工作流程。

在数据写入过程中,我们可以随时通过 EMR Serverless Spark 对 Paimon 表进行交互式查询。

  1. 创建 SQL Compute,配置Paimon相关参数。

spark.sql.extensions                org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
spark.sql.catalog.paimon            org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.paimon.metastore  dlf
spark.sql.catalog.paimon.warehouse  oss://test/warehouse

  1. 启动 SQL Compute,回到任务开发页面,进行交互式查询

示例1:

select * from `paimon`.`test_paimon_db`.`test_append_tbl` limit 10;

查询结果:

示例2:

select count(*) from `paimon`.`test_paimon_db`.`test_append_tbl` where dt='2024-06-04' and hh='12';

查询结果:


Serverless Spark 作业调度

Serverless Spark 还支持作业调度。我们将开发好的任务进行发布,然后创建工作流,编排任务并完成发布工作流。可以配置调度策略,实现定期调度。以下通过该模块实现 Paimon 表的 Compact 调度。

  1. 在“任务开发”页面,编写Paimon Compact SQL,完成发布。

CALL paimon.sys.compact (
  table => 'test_paimon_db.test_append_tbl',
  partitions => 'dt=\"2024-06-04\",hh=\"12\"',
  order_strategy => 'zorder',
  order_by => 'category'
)
;

  1. 在“任务编排”页面,创建工作流,添加节点,配置成刚发布的任务。


每个工作流节点都可以配置节点专用的引擎版本和Spark运行配置。

该节点我们配置以下Paimon配置。

spark.sql.extensions                org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
spark.sql.catalog.paimon            org.apache.paimon.spark.SparkCatalog
spark.sql.catalog.paimon.metastore  dlf
spark.sql.catalog.paimon.warehouse  oss://test/warehouse

  1. 手动运行工作流。

  1. 验证。在工作流的作业调度执行成功后,我们可以通过 SQL Compute 查询 Paimon 的 files 系统表,快速得到 Compact 前后文件的变化。


示例,查询 dt=2024-06-04/hh=12 分区的文件信息:

SELECT file_path, record_count, file_size_in_bytes FROM `paimon`.`test_paimon_db`.`test_append_tbl$files` where partition='[2024-06-04, 12]';

查询结果(Compact 前):

查询结果(Compact 后):


总结

本文演示了使用实时计算 Flink 版和 Serverless Spark 产品快速构建 Paimon 数据湖分析的流程,包括数据入湖 OSS、交互式查询,以及离线Compact。Serverless Spark完全兼容Paimon,通过内置的DLF的元数据实现了和其余云产品如实时计算Flink版的元数据互通,形成了完整的流批一体的解决方案。同时支持灵活的作业运行方式和参数配置,能够满足实时分析、生产调度等多项需求。




EMR Serverless Spark 在 2024年5月正式开启公测,在公测期间可以免费使用最高 100 CU 计算资源,欢迎试用。如果您在使用 EMR Serverless Spark 版的过程中遇到任何疑问,可钉钉扫描以下二维码加入钉钉群(群号:58570004119)咨询。

相关链接

  1. EMR Serverless Spark 版官网:https://www.aliyun.com/product/bigdata/serverlessspark
  2. 产品控制台:https://emr-next.console.aliyun.com/
  3. 产品文档:https://help.aliyun.com/zh/emr/emr-serverless-spark/
  4. 快速入门:https://help.aliyun.com/zh/emr/emr-serverless-spark/getting-started/get-started-with-sql-task-development



相关实践学习
数据湖构建DLF快速入门
本教程通过使⽤数据湖构建DLF产品对于淘宝用户行为样例数据的分析,介绍数据湖构建DLF产品的数据发现和数据探索功能。
快速掌握阿里云 E-MapReduce
E-MapReduce 是构建于阿里云 ECS 弹性虚拟机之上,利用开源大数据生态系统,包括 Hadoop、Spark、HBase,为用户提供集群、作业、数据等管理的一站式大数据处理分析服务。 本课程主要介绍阿里云 E-MapReduce 的使用方法。
目录
相关文章
|
6天前
|
分布式计算 运维 Serverless
EMR Serverless Spark服务和EMR Serverless StarRocks服务的比较
**EMR Serverless Spark** 以其出色的稳定性、高效性能、减轻运维负担及成本优化著称,适合大规模数据处理。**EMR Serverless StarRocks** 则以高速查询、存算分离架构和灵活扩缩容见长,侧重企业级功能。两者在不同应用场景中有各自优势,选择应基于具体需求。更多详情,参考阿里云官方资源。
|
27天前
|
消息中间件 Java 关系型数据库
实时计算 Flink版操作报错合集之从 PostgreSQL 读取数据并写入 Kafka 时,遇到 "initial slot snapshot too large" 的错误,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
923 0
|
27天前
|
存储 SQL 关系型数据库
实时计算 Flink版操作报错合集之按时间恢复时,报错:在尝试读取binlog时发现所需的binlog位置不再可用,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
716 0
|
17天前
|
分布式计算 运维 Serverless
EMR Serverless Spark PySpark流任务体验报告
阿里云EMR Serverless Spark是一款全托管的云原生大数据计算服务,旨在简化数据处理流程,降低运维成本。测评者通过EMR Serverless Spark提交PySpark流任务,体验了从环境准备、集群创建、网络连接到任务管理的全过程。通过这次测评,可以看出阿里云EMR Serverless Spark适合有一定技术基础的企业,尤其是需要高效处理大规模数据的场景,但新用户需要投入时间和精力学习和适应。
108 43
EMR Serverless Spark PySpark流任务体验报告
|
8天前
|
弹性计算 分布式计算 Serverless
全托管一站式大规模数据处理和分析Serverless平台 | EMR Serverless Spark 评测
【7月更文挑战第6天】全托管一站式大规模数据处理和分析Serverless平台 | EMR Serverless Spark 评测
|
20天前
|
SQL 存储 NoSQL
贝壳找房基于Flink+Paimon进行全量数据实时分组排序的实践
本文投稿自贝壳家装数仓团队,在结合家装业务场景下所探索出的一种基于 Flink+Paimon 的排序方案。这种方案可以在实时环境对全量数据进行准确的分组排序,同时减少对内存资源的消耗。在这一方案中,引入了“事件时间分段”的概念,以避免 Flink State 中冗余数据对排序结果的干扰,在保证排序结果准确性的同时,减少了对内存的消耗。并且基于数据湖组件 Paimon 的聚合模型和 Audit Log 数据在数据湖内构建了拉链表,为排序结果提供了灵活的历史数据基础。
28623 1
贝壳找房基于Flink+Paimon进行全量数据实时分组排序的实践
|
25天前
|
分布式计算 运维 数据挖掘
【评测有奖】参加 EMR Serverless Spark 产品评测,赢机械键盘、充电宝等礼品!
即日起至2024年7月18日,参加 EMR Serverless Spark 产品评测,赢机械键盘、充电宝等礼品!
95 7
【评测有奖】参加 EMR Serverless Spark 产品评测,赢机械键盘、充电宝等礼品!
|
5天前
|
数据采集 运维 Cloud Native
Flink+Paimon在阿里云大数据云原生运维数仓的实践
构建实时云原生运维数仓以提升大数据集群的运维能力,采用 Flink+Paimon 方案,解决资源审计、拓扑及趋势分析需求。
333 0
Flink+Paimon在阿里云大数据云原生运维数仓的实践
|
9天前
|
分布式计算 运维 Serverless
通过 EMR Serverless Spark 提交 PySpark 流任务
在大数据快速发展的时代,流式处理技术对于实时数据分析至关重要。EMR Serverless Spark提供了一个强大而可扩展的平台,它不仅简化了实时数据处理流程,还免去了服务器管理的烦恼,提升了效率。本文将指导您使用EMR Serverless Spark提交PySpark流式任务,展示其在流处理方面的易用性和可运维性。
47 0
通过 EMR Serverless Spark 提交 PySpark 流任务
|
12天前
|
分布式计算 Hadoop Serverless
数据处理的艺术:EMR Serverless Spark实践及应用体验
阿里云EMR Serverless Spark是基于Spark的全托管大数据处理平台,融合云原生弹性与自动化,提供任务全生命周期管理,让数据工程师专注数据分析。它内置高性能Fusion Engine,性能比开源Spark提升200%,并有成本优化的Celeborn服务。支持计算存储分离、OSS-HDFS兼容、DLF元数据管理,实现一站式的开发体验和Serverless资源管理。适用于数据报表、科学项目等场景,简化开发与运维流程。用户可通过阿里云控制台快速配置和体验EMR Serverless Spark服务。