E-MapReduce:Integrate Paimon with Flink

Prerequisites

You have created a Dataflow or custom cluster with Flink and Paimon selected. For more information, see Create a cluster.

Limitations

• E-MapReduce V3.46.0 and V5.17.0 do not support DLF catalogs and Hive catalogs.

• You can use Flink SQL to read from and write to Paimon on clusters that run E-MapReduce V3.46.0 to V3.50.X and E-MapReduce V5.12.0 to V5.16.X.

  Note

  For E-MapReduce V3.51.X and later and E-MapReduce V5.17.X and later, refer to the Apache Paimon documentation and configure the integration in your EMR cluster.

Procedure

Step 1: Configure dependencies

You can read from and write to Paimon by using a Filesystem catalog, a Hive catalog, or a DLF catalog. Configure the dependencies based on the method you choose.

cp /opt/apps/PAIMON/paimon-current/lib/flink/*.jar /opt/apps/FLINK/flink-current/lib/

cp /opt/apps/PAIMON/paimon-current/lib/flink/*.jar /opt/apps/FLINK/flink-current/lib/
cp /opt/apps/FLINK/flink-current/opt/catalogs/hive-2.3.6/*.jar /opt/apps/FLINK/flink-current/lib/

cp /opt/apps/PAIMON/paimon-current/lib/flink/*.jar /opt/apps/FLINK/flink-current/lib/
cp /opt/apps/PAIMON/paimon-current/lib/jackson/*.jar /opt/apps/FLINK/flink-current/lib/
cp /opt/apps/METASTORE/metastore-*/hive2/*.jar /opt/apps/FLINK/flink-current/lib/
cp /opt/apps/FLINK/flink-current/opt/catalogs/hive-2.3.6/*.jar /opt/apps/FLINK/flink-current/lib/

Step 2: Start the cluster

This topic uses session mode as an example. For other modes, see Basic usage.

Run the following command to start a detached YARN session:

yarn-session.sh --detached

Step 3: Create a catalog

Paimon stores data and metadata in a file system such as HDFS or in object storage such as OSS-HDFS. The warehouse parameter specifies the root path. If the specified warehouse path does not exist, Paimon automatically creates it. If the path already exists, you can use the catalog to access the existing tables in that path.

You can also synchronize metadata to Hive or DLF to allow other services to access Paimon data.

Step 4: Read and write Paimon via streaming

Run the following Flink SQL statements to create a table in the catalog, and then read from and write to it.

-- Set the execution mode to streaming.
SET 'execution.runtime-mode' = 'streaming';

-- Paimon requires you to set a checkpoint interval for streaming jobs.
SET 'execution.checkpointing.interval' = '10s';

-- Use the catalog created in the previous step.
USE CATALOG test_catalog;

-- Create and use a test database.
CREATE DATABASE test_db;
USE test_db;

-- Use datagen to generate random data.
CREATE TEMPORARY TABLE datagen_source (
    uuid int,
    kind int,
    price int
) WITH (
    'connector' = 'datagen',
    'fields.kind.min' = '0',
    'fields.kind.max' = '9',
    'rows-per-second' = '10'
);

-- Create a Paimon table.
CREATE TABLE test_tbl (
    uuid int,
    kind int,
    price int,
    PRIMARY KEY (uuid) NOT ENFORCED
);

-- Write data to the Paimon table.
INSERT INTO test_tbl SELECT * FROM datagen_source;

-- Read data from the table.
-- The preceding streaming write job runs concurrently.
-- Ensure your Flink cluster has sufficient resources (task slots) for both jobs. Otherwise, this query will not execute.
SELECT kind, SUM(price) FROM test_tbl GROUP BY kind;

Step 5: Run an OLAP query on Paimon

Run the following Flink SQL statements to perform an OLAP query on the created table.

-- Set the execution mode to batch.
RESET 'execution.checkpointing.interval';
SET 'execution.runtime-mode' = 'batch';

-- Use tableau mode to print results directly in the terminal.
SET 'sql-client.execution.result-mode' = 'tableau';

-- Query data in the table.
SELECT kind, SUM(price) FROM test_tbl GROUP BY kind;

Step 6: Clean up resources

After you stop the job, run the following Flink SQL statement to drop the created table.

DROP TABLE test_tbl;