Guosmilesmile commented on code in PR #13608:
URL: https://github.com/apache/iceberg/pull/13608#discussion_r2230803388
##########
docs/docs/flink-writes.md:
##########
@@ -396,3 +396,149 @@ To use SinkV2 based implementation, replace `FlinkSink`
with `IcebergSink` in th
- The `RANGE` distribution mode is not yet available for the `IcebergSink`
- When using `IcebergSink` use `uidSuffix` instead of the `uidPrefix`
+
+
+## Dynamic Iceberg Flink Sink
+
+Dynamic Flink Iceberg Sink allows:
+
+1. **Writing to any number of tables**
+ A single sink can dynamically route records to multiple Iceberg tables.
+
+2. **Dynamic table creation and updates**
+ Tables are created and updated based on user-defined routing logic.
+
+3. **Dynamic schema and partition evolution**
+ Table schemas and partition specs update during streaming execution.
+
+All configurations are controlled through the `DynamicRecord` class,
eliminating the need for Flink job restarts when requirements change.
+
+```java
+
+ DynamicIcebergSink.forInput(dataStream)
+ .generator((inputRecord, out) -> out.collect(
+ new DynamicRecord(
+ TableIdentifier.of("db", "table"),
+ "branch",
+ SCHEMA,
+ (RowData) inputRecord,
+ PartitionSpec.unpartitioned(),
+ DistributionMode.HASH,
+ 2)))
+ .catalogLoader(CatalogLoader.hive("hive", new Configuration(),
Map.of()))
+ .writeParallelism(10)
+ .immediateTableUpdate(true)
+ .append();
+```
+
+### Configuration Example
+
+```java
+DynamicIcebergSink.Builder<RowData> builder =
DynamicIcebergSink.forInput(inputStream);
+
+// Set common properties
+builder
+ .set("write.parquet.compression-codec", "gzip");
+
+// Set Dynamic Sink specific options
+builder
+ .writeParallelism(4)
+ .uidPrefix("dynamic-sink")
+ .cacheMaxSize(500)
+ .cacheRefreshMs(5000);
+
+// Add generator and append sink
+builder.generator(new CustomRecordGenerator());
+builder.append();
+```
+
+### Dynamic Routing Configuration
+
+Dynamic table routing can be customized by implementing the
`DynamicRecordGenerator` interface:
+
+```java
+public class CustomRecordGenerator implements DynamicRecordGenerator<RowData> {
+ @Override
+ public DynamicRecord generate(RowData row) {
+ DynamicRecord record = new DynamicRecord();
+ // Set table name based on business logic
+ TableIdentifier tableIdentifier = TableIdentifier.of(database,
tableName);
+ record.setTableIdentifier(tableIdentifier);
+ record.setData(row);
+ // Set the maximum number of parallel writers for a given
table/branch/schema/spec
+ record.writeParallelism(2);
+ return record;
+ }
+}
+
+// Set custom record generator when building the sink
+DynamicIcebergSink.Builder<RowData> builder =
DynamicIcebergSink.forInput(inputStream);
+builder.generator(new CustomRecordGenerator());
+// ... other config ...
+builder.append();
+```
+The user should provide a converter which converts the input record to a
DynamicRecord.
+We need the following information (DynamicRecord) for every record:
+
+| Property | Description
|
+|---------------------|-------------------------------------------------------------------------------------------|
+| `TableIdentifier` | The target table to which the record will be written.
|
+| `Branch` | The target branch for writing the record (optional).
|
+| `Schema` | The schema of the record.
|
+| `Spec` | The expected partitioning specification for the
record. |
+| `RowData` | The actual row data to be written.
|
+| `DistributionMode` | The distribution mode for writing the record
(currently supports NONE or HASH). |
+| `Parallelism` | The maximum number of parallel writers for a given
table/branch/schema/spec (WriteTarget). |
+| `upsertMode` | Overrides this table's write.upsert.enabled
(optional). |
+| `equalityFields` | The equality fields for the table(optional).
|
Review Comment:
Ok, fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
