Flink API

Use the Flink builders to connect a DataStream to a Paimon table. The sink integrates writer routing, checkpoints, and commits with Flink. Use RichCdcSinkBuilder when incoming records also carry schema changes.

Input or task	API
Flink Row values with a known schema	FlinkSinkBuilder.forRow
Flink internal RowData values	FlinkSinkBuilder.forRowData
Read a table as a DataStream<Row>	FlinkSourceBuilder.buildForRow
Ingest typed CDC records with schema evolution	RichCdcSinkBuilder

For SQL transformations, you can also convert between DataStream and the Table API. See DataStream API Integration.

Dependency

Maven dependency:

<dependency>
  <groupId>org.apache.paimon</groupId>
  <artifactId>paimon-flink-1.20</artifactId>
  <version>2.2-SNAPSHOT</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-api-java-bridge</artifactId>
  <version>1.20.0</version>
  <scope>provided</scope>
</dependency>

Or download the jar file:

Paimon Flink.

Match the Paimon artifact and Flink dependencies to the Flink version you run. See Flink installation for connector and Hadoop runtime setup.

Prepare a table

The read and write examples use my_db.my_table from the Java API setup, with columns f0 STRING and f1 INT, primary key f0, and two fixed buckets. Use the same warehouse in each example. In a cluster, choose a filesystem accessible to all workers instead of the local path shown here.

Write to Table

Declare the input's field names and types in the same order as the Paimon schema. The example updates Alice's value from 12 to 100. Enable checkpointing for an unbounded streaming source so that the sink can publish data as checkpoints complete.

import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.sink.FlinkSinkBuilder;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.Table;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

public class WriteToTable {

    public static void writeTo() throws Exception {
        // Create the Flink execution environment.
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60_000);

        // create a changelog DataStream
        DataStream<Row> input =
                env.fromElements(
                                Row.ofKind(RowKind.INSERT, "Alice", 12),
                                Row.ofKind(RowKind.INSERT, "Bob", 5),
                                Row.ofKind(RowKind.UPDATE_BEFORE, "Alice", 12),
                                Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100))
                        .returns(
                                Types.ROW_NAMED(
                                        new String[] {"f0", "f1"}, Types.STRING, Types.INT));

        // get table from catalog
        Options catalogOptions = new Options();
        catalogOptions.set("warehouse", "file:///tmp/paimon-api-warehouse");
        try (Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions)) {
            Table table = catalog.getTable(Identifier.create("my_db", "my_table"));

            DataType inputType =
                    DataTypes.ROW(
                            DataTypes.FIELD("f0", DataTypes.STRING()),
                            DataTypes.FIELD("f1", DataTypes.INT()));
            FlinkSinkBuilder builder = new FlinkSinkBuilder(table).forRow(input, inputType);

            builder.build();
            env.execute();
        }
    }
}

Use builder.parallelism(...) to set sink parallelism. Select builder.overwrite() only for an intentional overwrite; see write semantics.

Read from Table

This example performs a bounded read of the current table state. Use sourceBounded(false) for a continuous source and choose the appropriate streaming read mode.

import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.source.FlinkSourceBuilder;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.Table;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;

public class ReadFromTable {

    public static void readFrom() throws Exception {
        // Create the Flink execution environment.
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // get table from catalog
        Options catalogOptions = new Options();
        catalogOptions.set("warehouse", "file:///tmp/paimon-api-warehouse");
        try (Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions)) {
            Table table = catalog.getTable(Identifier.create("my_db", "my_table"));

            FlinkSourceBuilder builder = new FlinkSourceBuilder(table)
                    .env(env)
                    .sourceBounded(true);

            DataStream<Row> dataStream = builder.buildForRow();

            try (CloseableIterator<Row> rows = dataStream.executeAndCollect()) {
                rows.forEachRemaining(System.out::println);
            }
        }
    }
}

After running the write example against the fresh sample table, the bounded result contains Alice with value 100 and Bob with value 5. Output order is not guaranteed. A continuous read's row kinds depend on the table's changelog configuration.

Use projection, predicate, limit, and sourceParallelism on the source builder to customize the scan. Predicate field indexes refer to the table schema.

CDC ingestion with schema evolution

RichCdcRecord carries field names, types, and string-encoded values. RichCdcSinkBuilder uses the catalog loader to apply schema changes and write the records. This also supports adding columns to a partial-update table.

Create this separate table first, using a Flink SQL catalog pointing to the same warehouse:

CREATE DATABASE IF NOT EXISTS my_db;
CREATE TABLE my_db.cdc_orders (
    order_id BIGINT,
    price DOUBLE,
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH ('bucket' = '2');

The second record below introduces dt. Supply a serializable CatalogLoader so runtime operators can access the same catalog and evolve the schema.

import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogLoader;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.sink.cdc.RichCdcRecord;
import org.apache.paimon.flink.sink.cdc.RichCdcSinkBuilder;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.Table;
import org.apache.paimon.types.DataTypes;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.paimon.types.RowKind.INSERT;

public class WriteCdcToTable {

    public static void writeTo() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60_000);

        DataStream<RichCdcRecord> dataStream =
                env.fromElements(
                        RichCdcRecord.builder(INSERT)
                                .field("order_id", DataTypes.BIGINT(), "123")
                                .field("price", DataTypes.DOUBLE(), "62.2")
                                .build(),
                        // dt field will be added with schema evolution
                        RichCdcRecord.builder(INSERT)
                                .field("order_id", DataTypes.BIGINT(), "245")
                                .field("price", DataTypes.DOUBLE(), "82.1")
                                .field("dt", DataTypes.TIMESTAMP(), "2023-06-12 20:21:12")
                                .build());

        Identifier identifier = Identifier.create("my_db", "cdc_orders");
        Options catalogOptions = new Options();
        catalogOptions.set("warehouse", "file:///tmp/paimon-api-warehouse");
        CatalogLoader catalogLoader =
                () -> FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
        try (Catalog catalog = catalogLoader.load()) {
            Table table = catalog.getTable(identifier);

            new RichCdcSinkBuilder(table)
                    .forRichCdcRecord(dataStream)
                    .identifier(identifier)
                    .catalogLoader(catalogLoader)
                    .build();

            env.execute();
        }
    }
}