Apache Paimon is a real-time Data Lake solution based on stream and batch processing that combines efficient storage and flexible computation, designed for processing large-scale real-time and streaming data. Paimon provides comprehensive Catalog interface support for the Flink engine, allowing for easy management of real-time Data Lake metadata.

1. Usage restrictions #

• Paimon supports various metadata types, including file system-based and LAS Catalog (Hive). To manage metadata with LAS Catalog, you need to activate the LAS Catalog service.
• The Paimon connector is only supported in Flink 1.16-volcano engine versions and higher.

2. Usage steps #

2.1 Create a Filesystem Catalog #

Before creating a Catalog, you need to create the corresponding TOS bucket and Catalog folder. For related documentation, see Create a bucket and Create a folder.

Note: ensure Flink and TOS are in the same Region. Flink currently does not support cross-Region access to TOS Buckets.

CREATE CATALOG ${catalog_name}
WITH
  (
    'type' = 'paimon',
    'warehouse' = 'tos://${bucket_name}/${catalog_name}'
  );

• ${catalog_name}: A custom name for the Catalog.
• ${bucket_name}: The name of the TOS (Object Storage) bucket that stores Paimon data.

2.2 Create an LAS Catalog #

Prerequisites: Flink requires API access to synchronize LAS metadata. Requirements

1. Use API Key Management to create an AccessKey/AccessKeySecret for a sub-account (to be entered in hive-site.xml later).
2. Grant LASFullAccess permission to the sub-account in Permission Management.

2.2.1 Create and Authorize an LAS Data Directory

Refer to Data Directory Management to create a data directory for Paimon in the LAS Catalog product. Enter the data catalog name and storage location.

Note: The TOS bucket and directory must be created in advance and be consistent with the directory path used in Flink SQL.

In the LAS Catalog permission management module, grant permissions to the newly created Catalog. For instructions on assigning permissions, refer to Permission Management. You will use Flink for subsequent operations, such as creating database tables and writing data. Therefore, we recommend granting Flink developers Admin permissions for the Catalog. Ensure you can perform the following tasks.

2.2.2 Create an LAS Catalog in Flink

2.2.2.1 (Recommended) LAS Connector Mode

Note: This mode is not supported in Flink 1.16 and earlier versions. If required, use the method described in 2.2.2.2 (Not Recommended) Hive Compatibility Mode.

CREATE CATALOG paimon_las_catalog_117 WITH (
    'type'='paimon',
    'metastore'='hive',
    -- The parameter indicates a connection using the LAS Connector.
    'is-las' = 'true',
    -- LAS Region. Supported regions include cn-beijing, cn-shanghai, and cn-guangzhou. Different regions require different URIs.
    'hive.client.las.region.name' = 'cn-beijing',
    'hive.metastore.uris' = 'thrift://lakeformation.las.cn-beijing.ivolces.com:48869',
    'hive.hms.client.is.public.cloud' = 'true',
    'hive.client.las.ak' = '<YOUR ACCESS KEY>',
    'hive.client.las.sk' = '<YOUR ACCESS KEY SECRET>',
    'metastore.catalog.default' = '<YOUR LAS CATALOG NAME>',
    -- This is the Paimon TOS storage directory.
    'warehouse' = 'tos://<YOUR BUCKET>/<YOUR WAREHOUSE PATH>',
);

2.2.2.2 (Not Recommended) Hive Compatibility Mode

Note: This solution is designed for Hive compatibility only. For Flink 1.17 and later, we recommend using 2.2.2.1 (Recommended) LAS Connector Mode.

This mode stores metadata in the LAS Catalog's metadata storage service and table files in a specified file system path.

CREATE CATALOG my_hive WITH (
  'type' = 'paimon',
  'metastore' = 'hive',
  -- The Thrift interface address for the LAS metadata service. Be sure to modify the specific Region information.
  'uri' = 'thrift://lakeformation.las.cn-beijing.ivolces.com:48869',
  -- The Hive Conf directory, uploaded via the dependency files module in the Flink job development interface
  'hive-conf-dir' = '/opt/tiger/workdir', 
  -- Storage bucket directory address
  'warehouse' = 'tos://<bucket-name>/path/to/warehouse'
);

The above SQL statement creates a Paimon Catalog named my_hive. Its metadata is stored in LAS Catalog, and its Data Warehouse path is tos://<bucket-name>/path/to/warehouse.
Additionally, when creating a job, you need to upload the corresponding hive.xml file on the job development page.
The content of the hive.xml file is as follows:

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
  <property>
    <name>hive.server2.max.start.attempts</name>
    <value>5</value>
  </property>
  <!-- Configure the LAS catalog name -->
  <property>
    <name>metastore.catalog.default</name>
    <value>__LAS_CATALOG_NAME__</value>
  </property>
  <!-- Configure the LAS region name, such as cn-beijing, cn-shanghai, or cn-guangzhou -->
  <property>
    <name>hive.client.las.region.name</name>
    <value>cn-shanghai</value>
  </property>
  <!-- Configure the LAS metastore URI according to the region -->
  <property>
    <name>hive.metastore.uris</name>
    <value>thrift://lakeformation.las.cn-shanghai.ivolces.com:48869</value>
  </property>
  <property>
    <name>hive.hms.client.is.public.cloud</name>
    <value>true</value>
  </property>
  <!-- Configure the LAS access key and access key secret -->
  <property>
    <name>hive.client.las.ak</name>
    <value>__LAS_ACCESS_KEY__</value>
  </property>
  <property>
    <name>hive.client.las.sk</name>
    <value>__LAS_ACCESS_KEY_SECRET__</value>
  </property>
</configuration>

Be sure to enter the specific LAS Catalog and account AK/SK information. Then, upload the hive.xml file to the dependency files in the job development section.
Additionally, if you use Flink-1.16-volcano, the engine does not have a built-in LAS Catalog connector. You need to manually upload the following JAR packages to the dependency files:

flink-sql-connector-hive-las-formation-3_2.12-1.16-byted-connector-SNAPSHOT.jar

Unknown size

2.3 Reading from and Writing to a Paimon Catalog #

2.3.1 Creating a Database

Create a database in the Catalog to organize and manage tables.

CREATE DATABASE IF NOT EXISTS ${catalog_name}.${db_name};

• ${db_name}: A custom name for the database.

2.3.2 Create a Data Table

Create a table in the database to define its structure and related configurations.

CREATE TABLE IF NOT EXISTS `${catalog_name}`.`${db_name}`.`${table_name}` (
    word varchar, -- Example field
    cnt bigint,
    PRIMARY KEY (word) NOT ENFORCED
) WITH (
    'bucket' = '4',  -- Controls the number of buckets. It is recommended that each bucket stores about 1GB of data.
    'changelog-producer' = 'input' -- Generates changelogs for downstream stream reading
);

• ${table_name}: A custom name for the table.
• bucket: The number of buckets. We recommend storing approximately 1 GB of data per bucket.
• changelog-producer：
  • Set to input to generate changelogs based on new upstream data for downstream streaming reads. Refer to Changelog output mechanism for more details. If you do not need a changelog, select the none option to save storage and write resources.

2.3.3 Querying Paimon Tables

Paimon query SQL supports both stream and batch reading. Simply select the appropriate mode during Flink execution:

INSERT INTO `print_table`
SELECT * FROM `paimon_test`.`default`.`doc_result`;

2.3.4 Stream and Batch Read/Write

Flink INSERT statements support both stream and batch writing semantics. Simply select the corresponding execution mode during runtime:

INSERT INTO `paimon_test`.`default`.`doc_result`
select
  t.word,
  count(1)
from
  doc_source t
GROUP BY
  t.word;

2.3.5 Overwrite

In addition to INSERT statements, Paimon also supports batch overwrites for data tables, partitions, and more:

-- Overwrite non-partitioned tables
INSERT OVERWRITE my_table SELECT ...

-- Overwrite partitioned table
INSERT OVERWRITE my_table PARTITION (key1 = value1, key2 = value2, ...) SELECT ...