Realtime Compute for Apache Flink:Write data to and consume data from a Paimon table

Use the CTAS or CDAS statement to synchronize data and schema changes

For information about how to use the CREATE TABLE AS (CTAS) or CREATE DATABASE AS (CDAS) statement, see Manage Apache Paimon catalogs.

Use the INSERT INTO statement to insert or update data

• A primary key table supports all types of changes (INSERT, UPDATE_BEFORE, UPDATE_AFTER, and DELETE) as inputs. If specific written data records have the same primary keys, they are merged based on the merge engine configuration.

• An append-only table (without primary keys) supports only INSERT-type changes as inputs.

Use the INSERT OVERWRITE statement to overwrite data

Overwriting is the process of replacing old data with new data. The old data is removed and can no longer be accessed. You can use the INSERT OVERWRITE statement to partially or completely overwrite a Paimon table. In the following examples, a table named my_table is used.

• Overwrite all data in an unpartitioned table.

  INSERT OVERWRITE my_table SELECT ...;

• Overwrite a partition of the table, such as the dt=20240108,hh=06 partition.

  INSERT OVERWRITE my_table PARTITION (`dt` = '20240108', `hh` = '06') SELECT ...;

• Overwrite certain partitions of the table. Specify the partitions that you want to overwrite in the SELECT statement.

  INSERT OVERWRITE my_table SELECT ...;

• Overwrite all data in a partitioned table.

  INSERT OVERWRITE my_table /*+ OPTIONS('dynamic-partition-overwrite' = 'false') */ SELECT ...;

Use the DELETE statement to delete data

You can use the DELETE statement to delete data from a primary key table. You must execute the DELETE statement in a script.

-- Delete all data records whose currency field is 'UNKNOWN' from the my_table table. 
DELETE FROM my_table WHERE currency = 'UNKNOWN';

Ignore DELETE changes

If a primary key table receives a DELETE change, the corresponding data is deleted by default. To ignore the DELETE changes and prevent data deletion, use SQL hints to set the ignore-delete parameter to true.

Adjust sink parallelism

You can use SQL hints to manually change the parallelism of the sink operator. The following table describes the parameter.

In the following example, the parallelism of the sink operator is set to 10.

INSERT INTO t /*+ OPTIONS('sink.parallelism' = '10') */ SELECT * FROM s;

Configure a consumer offset

To consume data from a Paimon table based on a specific offset, use the following methods:

• If you want to consume only the incremental data after the start time of the deployment, use SQL hints to specify 'scan.mode' = 'latest'.

  SELECT * FROM t /*+ OPTIONS('scan.mode' = 'latest') */;

• If you want to consume only the incremental data after a specific point in time, use SQL hints to specify the scan.timestamp-millis parameter. The parameter specifies the number of milliseconds that have elapsed since the epoch time UTC 1970-01-01 00:00:00.

  SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;

• If you want to consume all written data from a specific point in time and subsequent incremental data, use the following methods:

  Note

  The data files read by the source operator may contain a small amount of data that was written before the specified point in time because Paimon compacts small files for downstream consumption. You can use a WHERE condition in the SQL statement to filter data based on your business requirements.

  • Configure the Specify source's start time parameter when you start the deployment on the GUI.

  • Use SQL hints to specify the scan.file-creation-time-millis parameter.

    SELECT * FROM t /*+ OPTIONS('scan.file-creation-time-millis' = '1678883047356') */;

• If you want to consume only the incremental data starting from a specific snapshot file, use SQL hints to specify the scan.snapshot-id parameter. The parameter specifies the ID of the snapshot file that you want to use.

  SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;

• If you want to consume all data contained in a specific snapshot file and subsequent incremental data, use SQL hints to include the 'scan.mode' = 'from-snapshot-full' configuration and specify the scan.snapshot-id parameter. The scan.snapshot-id parameter specifies the ID of the snapshot file that you want to use.

  SELECT * FROM t /*+ OPTIONS('scan.mode' = 'from-snapshot-full', 'scan.snapshot-id' = '1') */;

Specify a consumer ID

A consumer ID is used to record the consumer offset in a Paimon table.

• If you modify the compute logic in the SQL statements, the deployment topology may change and the consumer offset may fail to be recovered from the state saved by Realtime Compute for Apache Flink. In this case, you can specify a consumer ID to save the consumer offset in the metadata files of the Paimon table. This way, the consumer can resume from the previous offset even if the deployment is restarted in a stateless manner.

• If an expired snapshot file is deleted before it is consumed , an error occurs. You can specify a consumer ID to prevent deletion of expired snapshot files that are not consumed.

To specify a consumer ID for the source operator, set the consumer-id parameter to a string. The first time you specify a consumer ID, the consumer offset is determined as described in Configure a consumer offset. When you specify the same consumer ID in subsequent operations, the consumer can resume from the saved consumer offset of the Paimon table.

For example, you can use the following SQL statement to specify a consumer ID named test-id for the source operator. If you want to reset the consumer offset of a consumer ID, specify 'consumer.ignore-progress' = 'true'.

SELECT * FROM t /*+ OPTIONS('consumer-id' = 'test-id') */;

Consume results of the INSERT OVERWRITE statement

By default, changes resulted from the INSERT OVERWRITE statement are not consumed by downstream operators in streaming mode. If you want to consume the changes, use SQL hints to specify 'streaming-read-overwrite' = 'true'.

SELECT * FROM t /*+ OPTIONS('streaming-read-overwrite' = 'true') */;