简介
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 版实时入湖
- 创建 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
- 创建 Paimon 表，按日期和小时分区，其中 category 模拟为 1~10 的随机数，用作后续执行 zorder 的列。设置表属性
write-only为true，关闭自动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' ) ;
- 启动入湖作业，通过 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 表进行交互式查询。
- 创建 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
- 启动 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 调度。
- 在“任务开发”页面，编写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' ) ;
- 在“任务编排”页面，创建工作流，添加节点，配置成刚发布的任务。
每个工作流节点都可以配置节点专用的引擎版本和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
- 手动运行工作流。
- 验证。在工作流的作业调度执行成功后，我们可以通过 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）咨询。
相关链接
- EMR Serverless Spark 版官网：https://www.aliyun.com/product/bigdata/serverlessspark
- 产品控制台：https://emr-next.console.aliyun.com/
- 产品文档：https://help.aliyun.com/zh/emr/emr-serverless-spark/
- 快速入门：https://help.aliyun.com/zh/emr/emr-serverless-spark/getting-started/get-started-with-sql-task-development
