基于Flink、EMR Serverless Spark与Paimon构建流批一体数据湖分析

实时计算Flink版

阿里云实时计算Flink版是一种全托管Serverless的Flink云服务，是一站式开发运维管理平台，开箱即用，计费灵活。具备作业开发、数据调试、运行与监控、自动调优、智能诊断等全生命周期能力。更多信息，请参见什么是阿里云实时计算Flink版。

Apache Paimon

Apache Paimon是一种统一的数据湖存储格式，结合Flink和Spark构建了流批处理的实时湖仓一体架构。Paimon创新地将湖格式与LSM（Log-structured merge-tree）技术结合，使数据湖具备了实时流更新和完整的流处理能力。更多信息，请参见Apache Paimon。

步骤一：通过实时计算Flink创建Paimon Catalog

Paimon Catalog可以方便地管理同一个warehouse目录下的所有Paimon表，并与其它阿里云产品连通。创建并使用Paimon Catalog，详情请参见管理Paimon Catalog。

1. 登录实时计算控制台。

2. 单击目标工作空间操作列下的控制台。

3. 创建Paimon Catalog。

   通过UI方法添加DLF类型的Paimon Catalog，无需手动填写AccessKey信息。具体操作，请参见管理Paimon Catalog。

4. 创建Paimon表。

   在查询脚本文本编辑区域输入如下命令后，选中代码后单击运行。

   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'
   );

5. 创建流作业。

   1. 新建作业。

      1. 在左侧导航栏，选择数据开发 > ETL。

      2. 新建流作业，在新建作业草稿对话框中，填写作业配置信息。

         作业参数	说明
         文件名称	作业的名称。 说明 作业名称在当前项目中必须保持唯一。
         引擎版本	当前作业使用的Flink的引擎版本。引擎版本号含义、版本对应关系和生命周期重要时间点详情请参见引擎版本介绍。

      3. 单击创建。

   2. 编写代码。

      在新建的作业草稿中，输入以下代码，通过datagen源源不断生成数据写入Paimon表中。

      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;

   3. 单击部署，即可将数据发布至生产环境。

   4. 您可以在作业运维页面启动作业进入运行阶段，详情请参见作业启动。

步骤二：通过EMR Serverless Spark创建SQL会话

创建的SQL会话用于SQL开发和查询。有关会话的详细介绍，请参见会话管理。

1. 进入会话管理页面。

   1. 登录E-MapReduce控制台。

   2. 在左侧导航栏，选择EMR Serverless > Spark。

   3. 在Spark页面，单击目标工作空间名称。

   4. 在EMR Serverless Spark页面，单击左侧导航栏中的会话管理。

2. 在Serverless Spark工作空间的数据目录中，添加步骤一中创建的DLF Catalog。

3. 创建SQL会话。

   1. 在SQL 会话页签，单击创建 SQL 会话。

   2. 在创建SQL会话页面，配置以下信息，其余参数无需配置，然后单击创建。

      参数	说明
      名称	自定义SQL会话的名称。例如，paimon_compute。

   3. 单击操作列的启动。

步骤三：通过EMR Serverless Spark进行交互式查询或任务调度

EMR Serverless Spark提供了交互式查询和任务调度两种操作模式，以满足不同的使用需求。交互式查询适用于快速查询和调试，而任务调度则支持任务的开发、发布和运维，实现完整的生命周期管理。

在数据写入过程中，我们可以随时使用EMR Serverless Spark对Paimon表进行交互式查询，以便实时获取数据状态和执行快速分析。此外，通过发布开发好的任务并创建工作流，可以编排各项任务并完成工作流的发布。您可以配置调度策略，实现任务的定期调度，从而保证数据处理和分析的自动化与高效性。