通过实时计算Flink和Paimon实现流批一体

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

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

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

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

3. 创建Paimon Catalog。

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

   2. 单击，新建查询脚本。

   3. 填写SQL代码。

      Catalog完整配置如下所示。

      CREATE CATALOG `paimon` WITH (
        'type' = 'paimon',
        'metastore' = 'dlf',
        'warehouse' = '<warehouse>',
        'dlf.catalog.id' = '<dlf.catalog.id>',
        'dlf.catalog.accessKeyId' = '<dlf.catalog.accessKeyId>',
        'dlf.catalog.accessKeySecret' = '<dlf.catalog.accessKeySecret>',
        'dlf.catalog.endpoint' = '<dlf.catalog.endpoint>',
        'dlf.catalog.region' = '<dlf.catalog.region>',
      );

      配置项	说明	是否必填	备注
      paimon	Paimon Catalog名称。	是	请填写为自定义的英文名。
      type	Catalog类型。	是	固定值为paimon。
      metastore	元数据存储类型。	是	本文示例元数据存储类型选择dlf，通过DLF实现统一的元数据管理，实现多引擎无缝衔接。
      warehouse	配置数据仓库的实际位置。	是	请根据实际情况修改。
      dlf.catalog.id	DLF数据目录ID。	是	请在数据湖构建控制台上查看数据目录对应的ID，具体操作请参见数据目录。
      dlf.catalog.accessKeyId	访问DLF服务所需的Access Key ID。	是	获取方法请参见创建AccessKey。
      dlf.catalog.accessKeySecret	访问DLF服务所需的Access Key Secret。	是	获取方法请参见创建AccessKey。
      dlf.catalog.endpoint	DLF服务的Endpoint。	是	详情请参见已开通的地域和访问域名。 说明 如果Flink与DLF位于同一地域，则使用VPC网络Endpoint，否则使用公网Endpoint。
      dlf.catalog.region	DLF所在区域。	是	详情请参见已开通的地域和访问域名。 说明 请和dlf.catalog.endpoint选择的地域保持一致。

   4. 选择或创建Session集群。

      单击页面右下角的执行环境，选择对应版本的Session集群（VVR 8.0.4及以上引擎版本）。如果没有Session集群，请参见步骤一：创建Session集群。

   5. 选中目标代码片段后，单击代码行左侧的运行。

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. 单击新建。

      3. 单击空白的流作业草稿。

      4. 单击下一步。

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

         作业参数	说明
         文件名称	作业的名称。 说明 作业名称在当前项目中必须保持唯一。
         存储位置	指定该作业的存储位置。 您还可以在现有文件夹右侧，单击图标，新建子文件夹。
         引擎版本	当前作业使用的Flink的引擎版本。引擎版本号含义、版本对应关系和生命周期重要时间点详情请参见引擎版本介绍。

      6. 单击创建。

   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. 创建SQL会话。

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

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

      参数	说明
      名称	自定义SQL会话的名称。例如，paimon_compute。
      Spark配置	请填写以下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 <warehouse> spark.sql.catalog.paimon.dlf.catalog.id <dlf.catalog.id> 请根据您的实际情况替换以下信息： <warehouse>：配置数据仓库的实际位置，请根据实际情况修改。 <dlf.catalog.id>：DLF数据目录ID，请根据实际情况修改。

   3. 单击操作列的启动。

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

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

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