lintingbin2009 opened a new pull request, #5425:
URL: https://github.com/apache/iceberg/pull/5425
FlinkSink support dynamically changed schema. This commit only implements
the dynamic update schema on the FlinkSink side, and users need to implement
the update of RowData in their own code. The code below is my example code for
dynamically building RowData.
```
public class RowDataFlatMap extends RichFlatMapFunction<Map<String, String>,
RowData>
implements CheckpointListener {
private static final Logger LOG =
LoggerFactory.getLogger(RowDataFlatMap.class);
private final TableLoader tableLoader;
private DataTypeConverter dataTypeConverter;
private Table table;
public RowDataFlatMap(TableLoader tableLoader) {
this.tableLoader = tableLoader;
}
@Override
public void open(Configuration parameters) {
tableLoader.open();
table = tableLoader.loadTable();
dataTypeConverter = new DataTypeConverter(table.schema());
}
@Override
public void close() throws Exception {
tableLoader.close();
}
@Override
public void flatMap(Map<String, String> value, Collector<RowData> out) {
try {
out.collect(dataTypeConverter.map2RowData(value));
} catch (LogProcessException e) {
LOG.debug("exception type: {}, value: {}", e.getClass(), value);
}
}
@Override
public void notifyCheckpointComplete(long checkpointId) {
table.refresh();
if (!dataTypeConverter.isSameSchema(table.schema())) {
dataTypeConverter.updateSchema(table.schema());
}
}
}
```
--
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]
