Flink API

We do not recommend using programming API. Paimon is designed for SQL first, unless you are a professional Flink developer, even if you do, it can be very difficult.

We strongly recommend that you use Flink SQL or Spark SQL, or simply use SQL APIs in programs.

The following documents are not detailed and are for reference only.

Dependency

Maven dependency:

  1. <dependency>
  2.   <groupId>org.apache.paimon</groupId>
  3.   <artifactId>paimon-flink-1.20</artifactId>
  4.   <version>0.9.0</version>
  5. </dependency>
  7. <dependency>
  8.   <groupId>org.apache.flink</groupId>
  9.   <artifactId>flink-table-api-java-bridge</artifactId>
  10.   <version>1.20.0</version>
  11.   <scope>provided</scope>
  12. </dependency>

Or download the jar file: Paimon Flink.

Please choose your Flink version.

Paimon relies on Hadoop environment, you should add hadoop classpath or bundled jar.

Not only DataStream API, you can also read or write to Paimon tables by the conversion between DataStream and Table in Flink. See DataStream API Integration.

Write to Table

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.Identifier;
  3. import org.apache.paimon.flink.FlinkCatalogFactory;
  4. import org.apache.paimon.flink.sink.FlinkSinkBuilder;
  5. import org.apache.paimon.options.Options;
  6. import org.apache.paimon.table.Table;
  8. import org.apache.flink.api.common.typeinfo.Types;
  9. import org.apache.flink.streaming.api.datastream.DataStream;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import org.apache.flink.table.api.DataTypes;
  12. import org.apache.flink.table.types.DataType;
  13. import org.apache.flink.types.Row;
  14. import org.apache.flink.types.RowKind;
  16. public class WriteToTable {
  18.     public static void writeTo() throws Exception {
  19.         // create environments of both APIs
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  21.         // for CONTINUOUS_UNBOUNDED source, set checkpoint interval
  22.         // env.enableCheckpointing(60_000);
  24.         // create a changelog DataStream
  25.         DataStream<Row> input =
  26.                 env.fromElements(
  27.                                 Row.ofKind(RowKind.INSERT, "Alice", 12),
  28.                                 Row.ofKind(RowKind.INSERT, "Bob", 5),
  29.                                 Row.ofKind(RowKind.UPDATE_BEFORE, "Alice", 12),
  30.                                 Row.ofKind(RowKind.UPDATE_AFTER, "Alice", 100))
  31.                         .returns(
  32.                                 Types.ROW_NAMED(
  33.                                         new String[] {"name", "age"}, Types.STRING, Types.INT));
  35.         // get table from catalog
  36.         Options catalogOptions = new Options();
  37.         catalogOptions.set("warehouse", "/path/to/warehouse");
  38.         Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
  39.         Table table = catalog.getTable(Identifier.create("my_db", "T"));
  41.         DataType inputType =
  42.                 DataTypes.ROW(
  43.                         DataTypes.FIELD("name", DataTypes.STRING()),
  44.                         DataTypes.FIELD("age", DataTypes.INT()));
  45.         FlinkSinkBuilder builder = new FlinkSinkBuilder(table).forRow(input, inputType);
  47.         // set sink parallelism
  48.         // builder.parallelism(_your_parallelism)
  50.         // set overwrite mode
  51.         // builder.overwrite(...)
  53.         builder.build();
  54.         env.execute();
  55.     }
  56. }

Read from Table

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.Identifier;
  3. import org.apache.paimon.flink.FlinkCatalogFactory;
  4. import org.apache.paimon.flink.source.FlinkSourceBuilder;
  5. import org.apache.paimon.options.Options;
  6. import org.apache.paimon.table.Table;
  8. import org.apache.flink.streaming.api.datastream.DataStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.types.Row;
  12. public class ReadFromTable {
  14.     public static void readFrom() throws Exception {
  15.         // create environments of both APIs
  16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         // get table from catalog
  19.         Options catalogOptions = new Options();
  20.         catalogOptions.set("warehouse", "/path/to/warehouse");
  21.         Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
  22.         Table table = catalog.getTable(Identifier.create("my_db", "T"));
  24.         // table = table.copy(Collections.singletonMap("scan.file-creation-time-millis", "..."));
  26.         FlinkSourceBuilder builder = new FlinkSourceBuilder(table).env(env);
  28.         // builder.sourceBounded(true);
  29.         // builder.projection(...);
  30.         // builder.predicate(...);
  31.         // builder.limit(...);
  32.         // builder.sourceParallelism(...);
  34.         DataStream<Row> dataStream = builder.buildForRow();
  36.         // use this datastream
  37.         dataStream.executeAndCollect().forEachRemaining(System.out::println);
  39.         // prints:
  40.         // +I[Bob, 12]
  41.         // +I[Alice, 12]
  42.         // -U[Alice, 12]
  43.         // +U[Alice, 14]
  44.     }
  45. }

Cdc ingestion Table

Paimon supports ingest data into Paimon tables with schema evolution.

  • You can use Java API to write cdc records into Paimon Tables.
  • You can write records to Paimon’s partial-update table with adding columns dynamically.

Here is an example to use RichCdcSinkBuilder API:

  1. import org.apache.paimon.catalog.Catalog;
  2. import org.apache.paimon.catalog.CatalogContext;
  3. import org.apache.paimon.catalog.CatalogFactory;
  4. import org.apache.paimon.flink.FlinkCatalogFactory;
  5. import org.apache.paimon.catalog.Identifier;
  6. import org.apache.paimon.flink.sink.cdc.RichCdcRecord;
  7. import org.apache.paimon.flink.sink.cdc.RichCdcSinkBuilder;
  8. import org.apache.paimon.fs.Path;
  9. import org.apache.paimon.options.Options;
  10. import org.apache.paimon.schema.Schema;
  11. import org.apache.paimon.table.Table;
  12. import org.apache.paimon.types.DataTypes;
  14. import org.apache.flink.streaming.api.datastream.DataStream;
  15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  17. import static org.apache.paimon.types.RowKind.INSERT;
  19. public class WriteCdcToTable {
  21.     public static void writeTo() throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         // for CONTINUOUS_UNBOUNDED source, set checkpoint interval
  24.         // env.enableCheckpointing(60_000);
  26.         DataStream<RichCdcRecord> dataStream =
  27.                 env.fromElements(
  28.                         RichCdcRecord.builder(INSERT)
  29.                                 .field("order_id", DataTypes.BIGINT(), "123")
  30.                                 .field("price", DataTypes.DOUBLE(), "62.2")
  31.                                 .build(),
  32.                         // dt field will be added with schema evolution
  33.                         RichCdcRecord.builder(INSERT)
  34.                                 .field("order_id", DataTypes.BIGINT(), "245")
  35.                                 .field("price", DataTypes.DOUBLE(), "82.1")
  36.                                 .field("dt", DataTypes.TIMESTAMP(), "2023-06-12 20:21:12")
  37.                                 .build());
  39.         Identifier identifier = Identifier.create("my_db", "T");
  40.         Options catalogOptions = new Options();
  41.         catalogOptions.set("warehouse", "/path/to/warehouse");
  42.         Catalog.Loader catalogLoader = 
  43.                 () -> FlinkCatalogFactory.createPaimonCatalog(catalogOptions);
  44.         Table table = catalogLoader.load().getTable(identifier);
  46.         new RichCdcSinkBuilder(table)
  47.                 .forRichCdcRecord(dataStream)
  48.                 .identifier(identifier)
  49.                 .catalogLoader(catalogLoader)
  50.                 .build();
  52.         env.execute();
  53.     }
  54. }