E-MapReduce:Use Realtime Compute for Apache Flink and Paimon to implement unified batch and stream computing

Step 1: Create a Paimon catalog in Realtime Compute for Apache Flink

Apache Paimon catalogs can be used to manage all Apache Paimon tables in the same warehouse directory in an efficient manner. Apache Paimon catalogs can also be used by other Alibaba Cloud services. For information about how to create and use an Apache Paimon catalog, see Manage Apache Paimon catalogs.

1. Log on to the management console of Realtime Compute for Apache Flink.

2. Find the desired workspace and click Console in the Actions column.

3. Create an Apache Paimon catalog.

   1. In the left-side navigation pane, choose Development > Scripts.

   2. On the Scripts tab, click the icon to create a script.

   3. Enter the SQL code in the script editor.

      Sample code:

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

      Parameter	Description	Required	Remarks
      paimon	The name of the Apache Paimon catalog.	Yes	Enter a custom name.
      type	The type of the catalog.	Yes	Set the value to paimon.
      metastore	The metadata storage type.	Yes	Set the value to dlf. You can use DLF to manage metadata in a centralized manner and implement seamless integration among engines.
      warehouse	The data warehouse directory.	Yes	Configure this parameter based on your business requirements.
      dlf.catalog.id	The ID of the DLF data catalog.	Yes	You can view the ID of the data catalog in the DLF console.
      dlf.catalog.accessKeyId	The AccessKey ID that is used to access DLF.	Yes	For information about how to obtain the AccessKey pair, see Create an AccessKey pair.
      dlf.catalog.accessKeySecret	The AccessKey secret that is used to access DLF.	Yes	For information about how to obtain the AccessKey pair, see Create an AccessKey pair.
      dlf.catalog.endpoint	The endpoint of DLF.	Yes	For more information, see Supported regions and endpoints. Note If DLF resides in the same region as Realtime Compute for Apache Flink, the VPC endpoint is used. Otherwise, the public endpoint is used.
      dlf.catalog.region	The region in which DLF resides.	Yes	For more information, see Supported regions and endpoints. Note Make sure that the value of this parameter matches the endpoint that is specified by the dlf.catalog.endpoint parameter.

   4. Select or create a session cluster.

      In the lower-right corner of the script editing page, select a session cluster from the Environment drop-down list. You can select a session cluster that uses Ververica Runtime (VVR) 8.0.4 or later. If no session cluster is available, create a session cluster. For information about how to create a session cluster, see the Step 1: Create a session cluster section of the "Debug a deployment" topic.

   5. Select the code that you want to run and click Run to the left of the code.

4. Create an Apache Paimon table.

   In the script editor, enter and select the following code. Then, click Run in the upper-left corner of the script editor.

   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. Create a streaming draft.

   1. Create a streaming draft.

      1. In the left-side navigation pane, choose Development > ETL.

      2. On the Drafts tab, click the New (Create Draft) icon.

      3. On the SQL Scripts tab of the New Draft dialog box, click Blank Stream Draft.

      4. Click Next.

      5. In the New Draft dialog box, configure the parameters of the draft. The following table describes the parameters.

         Parameter	Description
         Name	The name of the draft that you want to create. Note The draft name must be unique in the current project.
         Location	The folder in which the code file of the draft is saved. You can also click the icon to the right of an existing folder to create a subfolder.
         Engine Version	The engine version of Flink that is used by the draft. For more information about engine versions, version mappings, and important time points in the lifecycle of each version, see Engine version.

      6. Click Create.

   2. Write code.

      In the draft editor, enter the following code to continuously generate data by using Datagen and write data to the Paimon table. Sample code:

      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. In the upper-right corner of the toolbar, click Deploy to deploy the draft to the production environment.

   4. Start the deployment on the Deployments page. For more information, see Start a deployment.

Step 2: Create an SQL session in EMR Serverless Spark

You must create an SQL session for SQL development and queries. For more information about sessions, see Manage sessions.

1. Go to the Compute page.

   1. Log on to the EMR console.

   2. In the left-side navigation pane, choose EMR Serverless > Spark.

   3. On the Spark page, click the name of the workspace that you want to manage.

   4. In the left-side navigation pane of the EMR Serverless Spark page, click Compute.

2. Create an SQL session.

   1. On the SQL Sessions tab, click Create SQL Session.

   2. On the Create SQL Session page, configure parameters and click Create. The following table describes the parameters.

      Parameter	Description
      Name	The name of the SQL session. Example: paimon_compute.
      Spark Configuration	Enter the following code to connect to 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> Configure the following parameters based on your business requirements: <warehouse>: the data warehouse directory. <dlf.catalog.id>: The ID of the DLF data catalog.

   3. Click Start in the Actions column.

Step 3: Perform interactive queries and schedule jobs in EMR Serverless Spark

EMR Serverless Spark allows you to perform interactive queries and schedule jobs in a workflow to meet different requirements. You can perform interactive queries to implement fast queries and debugging. You can also create a workflow to develop, publish, and maintain jobs. This helps you implement complete lifecycle management of jobs.

During the data writing process, you can use EMR Serverless Spark to perform interactive queries on the Paimon table at any time. This helps you obtain the status of data in real time and perform fast analysis on data. You can publish developed jobs and create a workflow based on the jobs to orchestrate the jobs and publish the workflow. You can configure scheduling policies to schedule jobs on a regular basis. This way, data can be processed and analyzed in an automated and efficient manner.