Reading with DataStream

Amoro supports reading data in Batch or Streaming mode through Java API.

Batch mode

Using Batch mode to read the full and incremental data in the FileStore.

  • Non-primary key tables support reading full data in batch mode, snapshot data with a specified snapshot-id or timestamp, and incremental data with a specified snapshot interval.
  • The primary key table temporarily only supports reading the current full amount and later CDC data.
  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  2. InternalCatalogBuilder catalogBuilder =
  3.     InternalCatalogBuilder
  4.         .builder()
  5.         .metastoreUrl("thrift://<url>:<port>/<catalog_name>");
  7. TableIdentifier tableId = TableIdentifier.of("catalog_name", "database_name", "test_table");
  8. AmoroTableLoader tableLoader = AmoroTableLoader.of(tableId, catalogBuilder);
  10. Map<String, String> properties = new HashMap<>();
  11. //  Default is true.
  12. properties.put("streaming", "false");
  14. DataStream<RowData> batch = 
  15.     FlinkSource.forRowData()
  16.         .env(env)
  17.         .tableLoader(tableLoader)
  18.         // The primary key table only supports reading the current full amount and later CDC data temporarily, without the properties parameter .
  19.         .properties(properties)
  20.         .build();
  22. // print All data read
  23. batch.print();
  25. // Submit and execute the task
  26. env.execute("Test Amoro Batch Read");

The map properties contain below keys, currently only valid for non-primary key tables:

Key Default Type Required Description
case-sensitive false Boolean No Case-sensitive
snapshot-id (none) Long No Read the full amount of data of the specified snapshot, only effective when streaming is false or not configured
as-of-timestamp (none) String No Read the last time less than the timestamp The full amount of snapshot data is valid only when streaming is false or not configured
start-snapshot-id (none) String No When streaming is false, end-snapshot-id needs to be used to read the two intervals Incremental data (snapshot1, snapshot2]. When streaming is true, read the incremental data after the snapshot, if not specified, read the incremental data after the current snapshot (not including the current one)
end-snapshot-id (none ) String No Need to cooperate with start-snapshot-id to read incremental data in two intervals (snapshot1, snapshot2]

Streaming mode

Amoro supports reading incremental data in FileStore or LogStore through Java API in Streaming mode

Streaming mode (LogStore)

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
  2. InternalCatalogBuilder catalogBuilder = 
  3.     InternalCatalogBuilder
  4.         .builder()
  5.         .metastoreUrl("thrift://<url>:<port>/<catalog_name>");
  7. TableIdentifier tableId = TableIdentifier.of("catalog_name", "database_name", "test_table");
  8. AmoroTableLoader tableLoader = AmoroTableLoader.of(tableId, catalogBuilder);
  10. AmoroTable table = AmoroUtils.load(tableLoader);
  11. // Read table All fields. If you only read some fields, you can construct the schema yourself, for example: 
  12. // Schema userSchema = new Schema(new ArrayList<Types.NestedField>() {{
  13. //   add(Types.NestedField.optional(0, "f_boolean", Types.BooleanType.get()));
  14. //   add(Types.NestedField.optional(1, "f_int", Types.IntegerType.get()));
  15. // }});
  16. Schema schema = table.schema();
  18. // -----------Hidden Kafka--------------
  19. LogKafkaSource source = LogKafkaSource.builder(schema, table.properties()).build();
  21. or
  23. // -----------Hidden Pulsar--------------
  24. LogPulsarSource source = LogPulsarSource.builder(schema, table.properties()).build();
  26. DataStream<RowData> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Log Source");
  28. // Print all the read data
  29. stream.print();
  31. // Submit and execute the task
  32. env.execute("Test Amoro Stream Read");

Streaming mode (FileStore)

  1. StreamExecutionEnvironment env = ...;
  2. InternalCatalogBuilder catalogBuilder = ...;
  3. TableIdentifier tableId = ...;
  4. AmoroTableLoader tableLoader = ...;
  6. Map<String, String> properties = new HashMap<>();
  7. // default is true 
  8. properties.put("streaming", "true");
  10. DataStream<RowData> stream = 
  11.     FlinkSource.forRowData()
  12.         .env(env)
  13.         .tableLoader(tableLoader)
  14.         // The primary key table only supports reading the current full amount and later CDC data for the time being, without the properties parameter
  15.         .properties(properties)
  16.         .build();
  18. // Print All read data 
  19. stream.print();
  21. // Submit and execute the task
  22. env.execute("Test Amoro Stream Read");
  24. StreamExecutionEnvironment env = ...; 
  25. InternalCatalogBuilder catalogBuilder = ...; 
  26. TableIdentifier tableId = ...; 
  27. AmoroTableLoader tableLoader = ...; 
  28. Map properties = new HashMap<>(); 
  29. // default is true properties.put("streaming", "true"); 
  30. DataStream stream = 
  31.     FlinkSource.forRowData() 
  32.         .env(env) 
  33.         .tableLoader(tableLoader) 
  34.         // The primary key table only supports reading the current full amount and later CDC data for the time being, without the properties parameter
  35.         .properties(properties) 
  36.         .build(); 
  38. // print All read data 
  39. stream.print(); 
  41. // Submit and execute the task 
  42. env.execute("Test Amoro Stream Read");

DataStream API supports reading primary key tables and non-primary key tables. The configuration items supported by properties can refer to Querying With SQL chapter Hint Option

Writing with DataStream

Amoro table supports writing data to LogStore or FileStore through Java API

Overwrite data

Amoro table currently Only supports the existing data in the dynamic Overwrite table of the non-primary key table

  1. DataStream<RowData> input = ...;
  2. InternalCatalogBuilder catalogBuilder = ...;
  3. TableIdentifier tableId = ...;
  4. AmoroTableLoader tableLoader = ...;
  6. TableSchema FLINK_SCHEMA = TableSchema.builder()
  7.     .field("id", DataTypes.INT())
  8.     .field("name", DataTypes.STRING())
  9.     .field("op_time", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
  10.     .build();
  12. FlinkSink
  13.     .forRowData(input)
  14.     .tableLoader(tableLoader)
  15.     .overwrite(true)
  16.     .flinkSchema(FLINK_SCHEMA)
  17.     .build();
  19. // Submit and execute the task
  20. env.execute("Test Amoro Overwrite");
  21. DataStream input = ...; InternalCatalogBuilder catalogBuilder = ...; TableIdentifier tableId = ...; AmoroTableLoader tableLoader = ...; TableSchema FLINK_SCHEMA = TableSchema.builder() .field("id", DataTypes.INT()) .field ("name", DataTypes.STRING()) .field("op_time", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE()) .build(); FlinkSink .forRowData(input) .tableLoader(tableLoader) .overwrite(true) .flinkSchema(FLINK_SCHEMA) .build(); // Submit and execute the task env.execute(“Test Amoro Overwrite”);

Appending data

For the Amoro table, it supports specifying to write data to FileStore or LogStore through Java API.

  1. DataStream<RowData> input = ...;
  2. InternalCatalogBuilder catalogBuilder = ...;
  3. TableIdentifier tableId = ...;
  4. AmoroTableLoader tableLoader = ...;
  6. TableSchema FLINK_SCHEMA = TableSchema.builder()
  7.     .field("id", DataTypes.INT())
  8.     .field("name", DataTypes.STRING())
  9.     .field("op_time", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
  10.     .build();
  12. AmoroTable table = AmoroUtils.loadAmoroTable(tableLoader);
  14. table.properties().put("arctic.emit.mode", "log,file");
  16. FlinkSink
  17.     .forRowData(input)
  18.     .table(table)
  19.     .tableLoader(tableLoader)
  20.     .flinkSchema(FLINK_SCHEMA)
  21.     .build();
  23. env.execute("Test Amoro Append");

The DataStream API supports writing to primary key tables and non-primary key tables. The configuration items supported by properties can refer to Writing With SQL chapter Hint Options

TIPS

arctic.emit.mode contains log, you need to configure log-store.enabled = true Enable Log Configuration

arctic.emit.mode When file is included, the primary key table will only be written to ChangeStore, and the non-primary key table will be directly written to BaseStore.