Guosmilesmile commented on code in PR #13608:
URL: https://github.com/apache/iceberg/pull/13608#discussion_r2230803777
##########
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).
|
+
+### Schema Update
+
+The dynamic sink maintains an LRU cache for both table metadata and incoming
schemas, with eviction based on size and time constraints. When a DynamicRecord
contains a schema that is incompatible with the current table schema, a schema
update is triggered. This update can occur either immediately or via a
centralized executor, depending on the immediateTableUpdate configuration.
While centralized updates reduce load on the Catalog, they may introduce
backpressure on the sink.
+
+Supported schema updates:
+
+- Adding new columns
+- Widening existing column types (e.g., Integer → Long, Float → Double)
+
+Unsupported schema updates:
+
+- Dropping columns
+- Renaming columns
+
+Dropping columns is avoided to prevent issues with late or out-of-order data,
as removed fields cannot be easily restored without data loss. Renaming is
unsupported because schema comparison is name-based, and renames would require
additional metadata or hints to resolve.
+
+
+#### Cache
+
+There are two caches here: the table metadata cache and the input schema
cache. The size of the table metadata cache is controlled by `cacheMaxSize`,
while the size of the input schema cache is controlled by
`inputSchemasPerTableCacheMaxSize`.
+
+The table metadata cache stores the metadata information of tables (such as
schema, partition specifications, etc.) to avoid frequent access to the Catalog.
+
+The input schema cache stores the input schema of each table along with the
compatibility resolution results between the input schema and the table schema.
+In the input schema cache, reuse the same DynamicRecord.schema instance if the
record schema is the same helps cache hits and improve performance.
Review Comment:
Right,replace it now
--
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]
