E-MapReduce:Build a stream-batch unified data lake with Flink, EMR Serverless Spark, and Paimon

Realtime Compute for Apache Flink

Alibaba Cloud Realtime Compute for Apache Flink is a fully managed, serverless cloud service for Apache Flink. It provides a ready-to-use one-stop platform for development, operations, and maintenance and offers flexible billing. The service includes full lifecycle capabilities such as job development, data debugging, job execution and monitoring, automatic tuning, and intelligent diagnostics. For more information, see What is Realtime Compute for Apache Flink?.

Apache Paimon

Apache Paimon is a unified data lake storage format that uses Flink and Spark to enable a real-time lakehouse architecture for stream-batch unification. Paimon innovatively combines lakehouse formats with Log-structured Merge-tree (LSM) technology, giving the data lake real-time streaming updates and full stream processing capabilities. For more information, see Apache Paimon.

Step 1: Create a Paimon catalog

A Paimon catalog lets you manage all Paimon tables within the same warehouse directory and connect them to other Alibaba Cloud products. For more information about how to create and use a Paimon catalog, see Manage a Paimon catalog.

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

2. In the Actions column of your target workspace, click Console.

3. Create a Paimon catalog.

   Add a Paimon catalog of the DLF type by using the UI. You do not need to enter AccessKey information manually. For detailed instructions, see Manage a Paimon catalog.

4. Create a Paimon table.

   In the Script editor, enter the following command. Then, select the code and click Run.

   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 stream job.

   1. Create a new job.

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

      2. Create a new stream job. In the New Blank Stream Draft dialog box, configure the job parameters.

         Parameter	Description
         Name	The name of the job. Note The job name must be unique within the current project.
         Engine Version	The Flink engine version for the current job. For more information about engine version numbers, version compatibility, and lifecycle milestones, see Engine versions.

      3. Click create.

   2. Write the code.

      In the new stream draft, enter the following code to continuously generate data using the datagen connector and write it to the Paimon table.

      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. Click Deploy to publish the job to the production environment.

   4. On the Deployments page, start the job. For more information, see Start a job.

Step 2: Create an SQL session

An SQL session supports SQL development and queries. For more information about sessions, see Session Manager.

1. Go to the Sessions 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 target workspace.

   4. On the EMR Serverless Spark page, click Sessions in the left-side navigation pane.

2. In the data catalog of your Serverless Spark workspace, add the DLF Catalog that you created in Step 1.

3. Create an SQL session.

   1. On the SQL Session tab, click Connect to SQL Session.

   2. On the Create SQL Session page, configure the following parameter, leave the other parameters at their defaults, and then click create.

      Parameter	Description
      Name	A custom name for the SQL session. For example, paimon_compute.

   3. In the Actions column, click START.

Step 3: Run interactive queries or task orchestration

EMR Serverless Spark offers two operational modes for different needs: interactive query and task orchestration. The interactive query mode is ideal for quick queries and debugging, while the task orchestration mode supports the full lifecycle of job development, publishing, and maintenance.

While data is being written, you can use EMR Serverless Spark to run an interactive query on the Paimon table at any time to check the real-time data status and perform quick analysis. In addition, you can publish developed jobs and create a workflow to orchestrate various tasks. You can configure scheduling policies to automate tasks, ensuring efficient and automated data processing and analysis.