You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "pvary (via GitHub)" <gi...@apache.org> on 2023/09/26 12:53:14 UTC
[GitHub] [iceberg] pvary opened a new pull request, #8653: Flink: new sink base on the unified sink API - WIP
pvary opened a new pull request, #8653:
URL: https://github.com/apache/iceberg/pull/8653
### Summary
The Flink community created a new Sink specification in [FLIP-143](https://cwiki.apache.org/confluence/display/FLINK/FLIP-143%3A+Unified+Sink+API) with the explicit goal to guarantee the unified handling of the bounded and unbounded data streams. Later it was enhanced in [FLIP-191](https://cwiki.apache.org/confluence/display/FLINK/FLIP-191%3A+Extend+unified+Sink+interface+to+support+small+file+compaction) so there is a well defined place to execute small files compaction. The deprecation of the old SinkFunction is postponed to somewhere around Flink 2.0 based on the discussion on the dev mailing list , so the migration is not extremely urgent, but having the possibility to use the PostCommitTopology to execute the compaction of the small files could provide immediate benefits for the users of the Iceberg-Flink integration.
### Previous work
There is an existing Iceberg PR #4904 for the Sink migration by Liwei Li (https://github.com/hililiwei) and Kyle Bendickson (https://github.com/kbendick) with the [related documentation](https://docs.google.com/document/d/1G4O6JidAoKgbIdy8Ts73OfG_KBEMpsW-LkXIb89I5k8/edit#heading=h.qqlw5ghn3vp7) which is authored by the same team. The discussion there is stuck, and the PR has been out of date for almost a year now. **The current proposal builds heavily on their work and wants to keep them as the co-authors for the proposed change.**
To start the discussion, I have created the following document.
https://docs.google.com/document/d/1K1M4wb9r_Tr-SDsUvqLyBaI5F14eRcqe3-ud6io0Da0/edit?usp=sharing
I propose the following timeline:
1. Review the design document
2. Update the PR
3. PR reviews
4. Merge the PR to the Iceberg source
5. Restart the discussion about the missing features in the Flink community by creating a FLIP
6. Discuss/review/merge the relevant Flink changes
7. Release the Flink changes
8. Create a PR in the Iceberg repo to start using the new Flink features
9. Merge the Iceberg PR
10. Be happy 😀
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#issuecomment-2362461718
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#discussion_r1345706466
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import java.util.Map;
+import java.util.UUID;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo;
+import org.apache.flink.streaming.api.connector.sink2.WithPostCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.WithPreCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.sink.committer.CommonCommitter;
+import org.apache.iceberg.flink.sink.committer.SinkV2Aggregator;
+import org.apache.iceberg.flink.sink.committer.SinkV2Committable;
+import org.apache.iceberg.flink.sink.committer.SinkV2CommittableSerializer;
+import org.apache.iceberg.flink.sink.committer.SinkV2Committer;
+import org.apache.iceberg.flink.sink.writer.IcebergStreamWriterMetrics;
+import org.apache.iceberg.flink.sink.writer.RowDataTaskWriterFactory;
+import org.apache.iceberg.flink.sink.writer.SinkV2Writer;
+
+/**
+ * Flink v2 sink offer different hooks to insert custom topologies into the sink. We will use the
+ * following:
+ *
+ * <ul>
+ * <li>{@link WithPreWriteTopology} which redistributes the data to the writers based on the
+ * {@link DistributionMode}
+ * <li>{@link org.apache.flink.api.connector.sink2.SinkWriter} which writes data files, and
+ * generates a {@link SinkV2Committable} to store the {@link
+ * org.apache.iceberg.io.WriteResult}
+ * <li>{@link WithPreCommitTopology} which we use to to place the {@link SinkV2Aggregator} which
+ * merges the individual {@link org.apache.flink.api.connector.sink2.SinkWriter}'s {@link
+ * org.apache.iceberg.io.WriteResult}s to a single {@link
+ * org.apache.iceberg.flink.sink.committer.DeltaManifests}
+ * <li>{@link Committer} which stores the incoming {@link
+ * org.apache.iceberg.flink.sink.committer.DeltaManifests}s in state for recovery, and commits
+ * them to the Iceberg table using the {@link CommonCommitter}
+ * <li>{@link WithPostCommitTopology} we could use for incremental compaction later. This is not
+ * implemented yet.
+ * </ul>
+ *
+ * The job graph looks like below:
+ *
+ * <pre>{@code
+ * Flink sink
+ * +-----------------------------------------------------------------------------------+
+ * | |
+ * +-------+ | +----------+ +-------------+ +---------------+ |
+ * | Map 1 | ==> | | writer 1 | | committer 1 | ---> | post commit 1 | |
+ * +-------+ | +----------+ +-------------+ +---------------+ |
+ * | \ / \ |
+ * | \ / \ |
+ * | \ / \ |
+ * +-------+ | +----------+ \ +-------------------+ / +-------------+ \ +---------------+ |
+ * | Map 2 | ==> | | writer 2 | --->| commit aggregator | | committer 2 | | post commit 2 | |
+ * +-------+ | +----------+ +-------------------+ +-------------+ +---------------+ |
+ * | Commit only on |
+ * | committer 1 |
+ * +-----------------------------------------------------------------------------------+
+ * }</pre>
+ */
+public class IcebergSink extends SinkBase
+ implements WithPreWriteTopology<RowData>,
+ WithPreCommitTopology<RowData, SinkV2Committable>,
+ WithPostCommitTopology<RowData, SinkV2Committable> {
+ private Table initTable;
+ private final TableLoader tableLoader;
+ private final Map<String, String> snapshotProperties;
+ private String uidPrefix;
+ private Committer<SinkV2Committable> sinkV2Committer;
+ private final String sinkId;
+ private CommonCommitter commonCommitter;
+
+ // FlinkWriteConf properties stored here for serialization
+ private boolean upsertMode;
+ private FileFormat dataFileFormat;
+ private long targetDataFileSize;
+
+ private IcebergSink(SinkBuilder builder) {
+ super(builder);
+
+ this.tableLoader = builder.tableLoader();
+ this.snapshotProperties = builder.snapshotProperties();
+
+ // We generate a random UUID every time when a sink is created.
+ // This is used to separate files generated by different sinks writing the same table.
+ // Also used to generate the aggregator operator name
+ this.sinkId = UUID.randomUUID().toString();
+ }
+
+ @Override
+ void init() {
+ super.init();
+ this.initTable = tableSupplier().get();
+ this.uidPrefix = builder().uidPrefix() == null ? "" : builder().uidPrefix();
+
+ // FlinkWriteConf properties needed to be set separately, so we do not have to serialize the
+ // full conf
+ this.upsertMode = flinkWriteConf().upsertMode();
+ this.dataFileFormat = flinkWriteConf().dataFileFormat();
+ this.targetDataFileSize = flinkWriteConf().targetDataFileSize();
+
+ this.commonCommitter =
+ new CommonCommitter(
+ tableLoader,
+ flinkWriteConf().branch(),
+ snapshotProperties,
+ flinkWriteConf().overwriteMode(),
+ flinkWriteConf().workerPoolSize(),
+ sinkId);
+ }
+
+ @Override
+ public DataStream<RowData> addPreWriteTopology(DataStream<RowData> inputDataStream) {
+ return distributeDataStream(inputDataStream);
+ }
+
+ @Override
+ public SinkV2Writer createWriter(InitContext context) {
+ RowDataTaskWriterFactory taskWriterFactory =
+ new RowDataTaskWriterFactory(
+ tableSupplier(),
+ flinkRowType(),
+ targetDataFileSize,
+ dataFileFormat,
+ writeProperties(),
+ equalityFieldIds(),
+ upsertMode);
+ IcebergStreamWriterMetrics metrics =
+ new IcebergStreamWriterMetrics(context.metricGroup(), initTable.name());
+ return new SinkV2Writer(
+ tableSupplier(),
+ taskWriterFactory,
+ metrics,
+ context.getSubtaskId(),
+ context.getAttemptNumber());
+ }
+
+ @Override
+ public DataStream<CommittableMessage<SinkV2Committable>> addPreCommitTopology(
+ DataStream<CommittableMessage<SinkV2Committable>> writeResults) {
+ TypeInformation<CommittableMessage<SinkV2Committable>> typeInformation =
+ CommittableMessageTypeInfo.of(this::getCommittableSerializer);
+ return writeResults
+ .global()
Review Comment:
We don't really need `global()` as the parallelism is forced to be 1
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#issuecomment-2378157962
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#discussion_r1345705474
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import java.util.Map;
+import java.util.UUID;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo;
+import org.apache.flink.streaming.api.connector.sink2.WithPostCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.WithPreCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.sink.committer.CommonCommitter;
+import org.apache.iceberg.flink.sink.committer.SinkV2Aggregator;
+import org.apache.iceberg.flink.sink.committer.SinkV2Committable;
+import org.apache.iceberg.flink.sink.committer.SinkV2CommittableSerializer;
+import org.apache.iceberg.flink.sink.committer.SinkV2Committer;
+import org.apache.iceberg.flink.sink.writer.IcebergStreamWriterMetrics;
+import org.apache.iceberg.flink.sink.writer.RowDataTaskWriterFactory;
+import org.apache.iceberg.flink.sink.writer.SinkV2Writer;
+
+/**
+ * Flink v2 sink offer different hooks to insert custom topologies into the sink. We will use the
+ * following:
+ *
+ * <ul>
+ * <li>{@link WithPreWriteTopology} which redistributes the data to the writers based on the
+ * {@link DistributionMode}
+ * <li>{@link org.apache.flink.api.connector.sink2.SinkWriter} which writes data files, and
+ * generates a {@link SinkV2Committable} to store the {@link
+ * org.apache.iceberg.io.WriteResult}
+ * <li>{@link WithPreCommitTopology} which we use to to place the {@link SinkV2Aggregator} which
+ * merges the individual {@link org.apache.flink.api.connector.sink2.SinkWriter}'s {@link
+ * org.apache.iceberg.io.WriteResult}s to a single {@link
+ * org.apache.iceberg.flink.sink.committer.DeltaManifests}
+ * <li>{@link Committer} which stores the incoming {@link
+ * org.apache.iceberg.flink.sink.committer.DeltaManifests}s in state for recovery, and commits
+ * them to the Iceberg table using the {@link CommonCommitter}
+ * <li>{@link WithPostCommitTopology} we could use for incremental compaction later. This is not
+ * implemented yet.
+ * </ul>
+ *
+ * The job graph looks like below:
+ *
+ * <pre>{@code
+ * Flink sink
+ * +-----------------------------------------------------------------------------------+
+ * | |
+ * +-------+ | +----------+ +-------------+ +---------------+ |
+ * | Map 1 | ==> | | writer 1 | | committer 1 | ---> | post commit 1 | |
+ * +-------+ | +----------+ +-------------+ +---------------+ |
+ * | \ / \ |
+ * | \ / \ |
+ * | \ / \ |
+ * +-------+ | +----------+ \ +-------------------+ / +-------------+ \ +---------------+ |
+ * | Map 2 | ==> | | writer 2 | --->| commit aggregator | | committer 2 | | post commit 2 | |
+ * +-------+ | +----------+ +-------------------+ +-------------+ +---------------+ |
+ * | Commit only on |
+ * | committer 1 |
+ * +-----------------------------------------------------------------------------------+
+ * }</pre>
+ */
+public class IcebergSink extends SinkBase
+ implements WithPreWriteTopology<RowData>,
+ WithPreCommitTopology<RowData, SinkV2Committable>,
+ WithPostCommitTopology<RowData, SinkV2Committable> {
+ private Table initTable;
+ private final TableLoader tableLoader;
+ private final Map<String, String> snapshotProperties;
+ private String uidPrefix;
+ private Committer<SinkV2Committable> sinkV2Committer;
+ private final String sinkId;
+ private CommonCommitter commonCommitter;
+
+ // FlinkWriteConf properties stored here for serialization
+ private boolean upsertMode;
+ private FileFormat dataFileFormat;
+ private long targetDataFileSize;
+
+ private IcebergSink(SinkBuilder builder) {
+ super(builder);
+
+ this.tableLoader = builder.tableLoader();
+ this.snapshotProperties = builder.snapshotProperties();
+
+ // We generate a random UUID every time when a sink is created.
+ // This is used to separate files generated by different sinks writing the same table.
+ // Also used to generate the aggregator operator name
+ this.sinkId = UUID.randomUUID().toString();
+ }
+
+ @Override
+ void init() {
+ super.init();
+ this.initTable = tableSupplier().get();
+ this.uidPrefix = builder().uidPrefix() == null ? "" : builder().uidPrefix();
+
+ // FlinkWriteConf properties needed to be set separately, so we do not have to serialize the
+ // full conf
+ this.upsertMode = flinkWriteConf().upsertMode();
+ this.dataFileFormat = flinkWriteConf().dataFileFormat();
+ this.targetDataFileSize = flinkWriteConf().targetDataFileSize();
+
+ this.commonCommitter =
+ new CommonCommitter(
+ tableLoader,
+ flinkWriteConf().branch(),
+ snapshotProperties,
+ flinkWriteConf().overwriteMode(),
+ flinkWriteConf().workerPoolSize(),
+ sinkId);
+ }
+
+ @Override
+ public DataStream<RowData> addPreWriteTopology(DataStream<RowData> inputDataStream) {
+ return distributeDataStream(inputDataStream);
+ }
+
+ @Override
+ public SinkV2Writer createWriter(InitContext context) {
+ RowDataTaskWriterFactory taskWriterFactory =
+ new RowDataTaskWriterFactory(
+ tableSupplier(),
+ flinkRowType(),
+ targetDataFileSize,
+ dataFileFormat,
+ writeProperties(),
+ equalityFieldIds(),
+ upsertMode);
+ IcebergStreamWriterMetrics metrics =
+ new IcebergStreamWriterMetrics(context.metricGroup(), initTable.name());
+ return new SinkV2Writer(
+ tableSupplier(),
+ taskWriterFactory,
+ metrics,
+ context.getSubtaskId(),
+ context.getAttemptNumber());
+ }
+
+ @Override
+ public DataStream<CommittableMessage<SinkV2Committable>> addPreCommitTopology(
+ DataStream<CommittableMessage<SinkV2Committable>> writeResults) {
+ TypeInformation<CommittableMessage<SinkV2Committable>> typeInformation =
+ CommittableMessageTypeInfo.of(this::getCommittableSerializer);
+ return writeResults
+ .global()
+ .transform(
+ prefixIfNotNull(uidPrefix, initTable.name() + "-" + sinkId + "-pre-commit-topology"),
+ typeInformation,
+ new SinkV2Aggregator(commonCommitter))
+ .uid(prefixIfNotNull(uidPrefix, "pre-commit-topology"))
Review Comment:
Maybe a better name/uid would be `...-write-result-aggregator` in case we add more things here later and it's also more descriptive
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] pvary commented on pull request #8653: Flink: new sink base on the unified sink API - WIP
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#issuecomment-1735565153
CC: @hililiwei, @stevenzwu, @chenjunjiedada, @gyfora - you might be interested in this
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#discussion_r1345500193
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:
##########
@@ -18,72 +18,39 @@
*/
package org.apache.iceberg.flink.sink;
-import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION;
-import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION_LEVEL;
-import static org.apache.iceberg.TableProperties.ORC_COMPRESSION;
-import static org.apache.iceberg.TableProperties.ORC_COMPRESSION_STRATEGY;
-import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION;
-import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_LEVEL;
-import static org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
-
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.time.Duration;
import java.util.List;
import java.util.Map;
-import java.util.Set;
-import java.util.function.Function;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.util.DataFormatConverters;
-import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.Row;
-import org.apache.iceberg.DataFile;
-import org.apache.iceberg.DistributionMode;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.PartitionField;
-import org.apache.iceberg.PartitionSpec;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.SerializableTable;
import org.apache.iceberg.Table;
-import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.flink.FlinkWriteConf;
-import org.apache.iceberg.flink.FlinkWriteOptions;
-import org.apache.iceberg.flink.TableLoader;
-import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
+import org.apache.iceberg.flink.sink.committer.IcebergFilesCommitter;
+import org.apache.iceberg.flink.sink.writer.IcebergStreamWriter;
import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
-import org.apache.iceberg.relocated.com.google.common.collect.Sets;
-import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.util.SerializableSupplier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class FlinkSink {
- private static final Logger LOG = LoggerFactory.getLogger(FlinkSink.class);
+public class FlinkSink extends SinkBase {
Review Comment:
Isn't this going to cause backward compatibility issues with the current FlinkSink instances? Maybe I am wrong but it means that users can only upgrade the iceberg version if they re-build their job graph (re-serialize operators etc.)
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/committer/CommonCommitter.java:
##########
@@ -0,0 +1,426 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink.committer;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.core.io.SimpleVersionedSerialization;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ReplacePartitions;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotUpdate;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.ThreadPools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Encapsulates the common commit related tasks which are needed for both the {@link
+ * org.apache.iceberg.flink.sink.FlinkSink} (SinkV1) and the {@link
+ * org.apache.iceberg.flink.sink.IcebergSink} (SinkV2).
+ *
+ * <p>The object is serialized and sent to the {@link IcebergFilesCommitter} (SinkV1) and to the
+ * {@link SinkV2Aggregator}, {@link SinkV2Committer} (SinkV2) to execute the related tasks. Before
+ * calling any of the public methods, the {@link #init(IcebergFilesCommitterMetrics)} method should
+ * be called, so the non-serializable attributes are initialized. The {@link
+ * #init(IcebergFilesCommitterMetrics)} is idempotent, so it could be called multiple times to
+ * accommodate the missing init feature in the {@link Committer}.
+ */
+public class CommonCommitter implements Serializable {
Review Comment:
Similar to the Sink base class maybe this should be called `CommitterBase` ?
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java:
##########
@@ -91,6 +91,12 @@ private FlinkConfigOptions() {}
.defaultValue(false)
.withDescription("Use the FLIP-27 based Iceberg source implementation.");
+ public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_USE_FLIP143_SINK =
Review Comment:
I think this config name is not very good. We should not include "flip" in it.
How about calling `use-v2-sink`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #8653: Flink: new sink base on the unified sink API - WIP
URL: https://github.com/apache/iceberg/pull/8653
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#discussion_r1345702939
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import java.util.Map;
+import java.util.UUID;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo;
+import org.apache.flink.streaming.api.connector.sink2.WithPostCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.WithPreCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.sink.committer.CommonCommitter;
+import org.apache.iceberg.flink.sink.committer.SinkV2Aggregator;
+import org.apache.iceberg.flink.sink.committer.SinkV2Committable;
+import org.apache.iceberg.flink.sink.committer.SinkV2CommittableSerializer;
+import org.apache.iceberg.flink.sink.committer.SinkV2Committer;
+import org.apache.iceberg.flink.sink.writer.IcebergStreamWriterMetrics;
+import org.apache.iceberg.flink.sink.writer.RowDataTaskWriterFactory;
+import org.apache.iceberg.flink.sink.writer.SinkV2Writer;
+
+/**
+ * Flink v2 sink offer different hooks to insert custom topologies into the sink. We will use the
+ * following:
+ *
+ * <ul>
+ * <li>{@link WithPreWriteTopology} which redistributes the data to the writers based on the
+ * {@link DistributionMode}
+ * <li>{@link org.apache.flink.api.connector.sink2.SinkWriter} which writes data files, and
+ * generates a {@link SinkV2Committable} to store the {@link
+ * org.apache.iceberg.io.WriteResult}
+ * <li>{@link WithPreCommitTopology} which we use to to place the {@link SinkV2Aggregator} which
+ * merges the individual {@link org.apache.flink.api.connector.sink2.SinkWriter}'s {@link
+ * org.apache.iceberg.io.WriteResult}s to a single {@link
+ * org.apache.iceberg.flink.sink.committer.DeltaManifests}
+ * <li>{@link Committer} which stores the incoming {@link
+ * org.apache.iceberg.flink.sink.committer.DeltaManifests}s in state for recovery, and commits
+ * them to the Iceberg table using the {@link CommonCommitter}
+ * <li>{@link WithPostCommitTopology} we could use for incremental compaction later. This is not
+ * implemented yet.
+ * </ul>
+ *
+ * The job graph looks like below:
+ *
+ * <pre>{@code
+ * Flink sink
+ * +-----------------------------------------------------------------------------------+
+ * | |
+ * +-------+ | +----------+ +-------------+ +---------------+ |
+ * | Map 1 | ==> | | writer 1 | | committer 1 | ---> | post commit 1 | |
+ * +-------+ | +----------+ +-------------+ +---------------+ |
+ * | \ / \ |
+ * | \ / \ |
+ * | \ / \ |
+ * +-------+ | +----------+ \ +-------------------+ / +-------------+ \ +---------------+ |
+ * | Map 2 | ==> | | writer 2 | --->| commit aggregator | | committer 2 | | post commit 2 | |
+ * +-------+ | +----------+ +-------------------+ +-------------+ +---------------+ |
+ * | Commit only on |
+ * | committer 1 |
+ * +-----------------------------------------------------------------------------------+
+ * }</pre>
+ */
+public class IcebergSink extends SinkBase
+ implements WithPreWriteTopology<RowData>,
+ WithPreCommitTopology<RowData, SinkV2Committable>,
+ WithPostCommitTopology<RowData, SinkV2Committable> {
+ private Table initTable;
+ private final TableLoader tableLoader;
+ private final Map<String, String> snapshotProperties;
+ private String uidPrefix;
+ private Committer<SinkV2Committable> sinkV2Committer;
+ private final String sinkId;
+ private CommonCommitter commonCommitter;
+
+ // FlinkWriteConf properties stored here for serialization
+ private boolean upsertMode;
+ private FileFormat dataFileFormat;
+ private long targetDataFileSize;
+
+ private IcebergSink(SinkBuilder builder) {
+ super(builder);
+
+ this.tableLoader = builder.tableLoader();
+ this.snapshotProperties = builder.snapshotProperties();
+
+ // We generate a random UUID every time when a sink is created.
+ // This is used to separate files generated by different sinks writing the same table.
+ // Also used to generate the aggregator operator name
+ this.sinkId = UUID.randomUUID().toString();
+ }
+
+ @Override
+ void init() {
+ super.init();
+ this.initTable = tableSupplier().get();
+ this.uidPrefix = builder().uidPrefix() == null ? "" : builder().uidPrefix();
+
+ // FlinkWriteConf properties needed to be set separately, so we do not have to serialize the
+ // full conf
+ this.upsertMode = flinkWriteConf().upsertMode();
+ this.dataFileFormat = flinkWriteConf().dataFileFormat();
+ this.targetDataFileSize = flinkWriteConf().targetDataFileSize();
+
+ this.commonCommitter =
+ new CommonCommitter(
+ tableLoader,
+ flinkWriteConf().branch(),
+ snapshotProperties,
+ flinkWriteConf().overwriteMode(),
+ flinkWriteConf().workerPoolSize(),
+ sinkId);
+ }
+
+ @Override
+ public DataStream<RowData> addPreWriteTopology(DataStream<RowData> inputDataStream) {
+ return distributeDataStream(inputDataStream);
+ }
+
+ @Override
+ public SinkV2Writer createWriter(InitContext context) {
+ RowDataTaskWriterFactory taskWriterFactory =
+ new RowDataTaskWriterFactory(
+ tableSupplier(),
+ flinkRowType(),
+ targetDataFileSize,
+ dataFileFormat,
+ writeProperties(),
+ equalityFieldIds(),
+ upsertMode);
+ IcebergStreamWriterMetrics metrics =
+ new IcebergStreamWriterMetrics(context.metricGroup(), initTable.name());
+ return new SinkV2Writer(
+ tableSupplier(),
+ taskWriterFactory,
+ metrics,
+ context.getSubtaskId(),
+ context.getAttemptNumber());
+ }
+
+ @Override
+ public DataStream<CommittableMessage<SinkV2Committable>> addPreCommitTopology(
+ DataStream<CommittableMessage<SinkV2Committable>> writeResults) {
+ TypeInformation<CommittableMessage<SinkV2Committable>> typeInformation =
+ CommittableMessageTypeInfo.of(this::getCommittableSerializer);
+ return writeResults
+ .global()
+ .transform(
+ prefixIfNotNull(uidPrefix, initTable.name() + "-" + sinkId + "-pre-commit-topology"),
+ typeInformation,
+ new SinkV2Aggregator(commonCommitter))
+ .uid(prefixIfNotNull(uidPrefix, "pre-commit-topology"))
+ .setParallelism(1)
+ .setMaxParallelism(1)
+ .global();
Review Comment:
SetP/MaxP could be replaced by `.forceNonParallel()`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#discussion_r1345500193
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java:
##########
@@ -18,72 +18,39 @@
*/
package org.apache.iceberg.flink.sink;
-import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION;
-import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION_LEVEL;
-import static org.apache.iceberg.TableProperties.ORC_COMPRESSION;
-import static org.apache.iceberg.TableProperties.ORC_COMPRESSION_STRATEGY;
-import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION;
-import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_LEVEL;
-import static org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
-
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.time.Duration;
import java.util.List;
import java.util.Map;
-import java.util.Set;
-import java.util.function.Function;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.util.DataFormatConverters;
-import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.Row;
-import org.apache.iceberg.DataFile;
-import org.apache.iceberg.DistributionMode;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.PartitionField;
-import org.apache.iceberg.PartitionSpec;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.SerializableTable;
import org.apache.iceberg.Table;
-import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.flink.FlinkWriteConf;
-import org.apache.iceberg.flink.FlinkWriteOptions;
-import org.apache.iceberg.flink.TableLoader;
-import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
+import org.apache.iceberg.flink.sink.committer.IcebergFilesCommitter;
+import org.apache.iceberg.flink.sink.writer.IcebergStreamWriter;
import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
-import org.apache.iceberg.relocated.com.google.common.collect.Sets;
-import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.util.SerializableSupplier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class FlinkSink {
- private static final Logger LOG = LoggerFactory.getLogger(FlinkSink.class);
+public class FlinkSink extends SinkBase {
Review Comment:
Isn't this going to cause backward compatibility issues with the current FlinkSink instances? Maybe I am wrong but it means that users can only upgrade the iceberg version if they re-build their job graph (re-serialize operators etc.)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#discussion_r1345702939
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import java.util.Map;
+import java.util.UUID;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo;
+import org.apache.flink.streaming.api.connector.sink2.WithPostCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.WithPreCommitTopology;
+import org.apache.flink.streaming.api.connector.sink2.WithPreWriteTopology;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.sink.committer.CommonCommitter;
+import org.apache.iceberg.flink.sink.committer.SinkV2Aggregator;
+import org.apache.iceberg.flink.sink.committer.SinkV2Committable;
+import org.apache.iceberg.flink.sink.committer.SinkV2CommittableSerializer;
+import org.apache.iceberg.flink.sink.committer.SinkV2Committer;
+import org.apache.iceberg.flink.sink.writer.IcebergStreamWriterMetrics;
+import org.apache.iceberg.flink.sink.writer.RowDataTaskWriterFactory;
+import org.apache.iceberg.flink.sink.writer.SinkV2Writer;
+
+/**
+ * Flink v2 sink offer different hooks to insert custom topologies into the sink. We will use the
+ * following:
+ *
+ * <ul>
+ * <li>{@link WithPreWriteTopology} which redistributes the data to the writers based on the
+ * {@link DistributionMode}
+ * <li>{@link org.apache.flink.api.connector.sink2.SinkWriter} which writes data files, and
+ * generates a {@link SinkV2Committable} to store the {@link
+ * org.apache.iceberg.io.WriteResult}
+ * <li>{@link WithPreCommitTopology} which we use to to place the {@link SinkV2Aggregator} which
+ * merges the individual {@link org.apache.flink.api.connector.sink2.SinkWriter}'s {@link
+ * org.apache.iceberg.io.WriteResult}s to a single {@link
+ * org.apache.iceberg.flink.sink.committer.DeltaManifests}
+ * <li>{@link Committer} which stores the incoming {@link
+ * org.apache.iceberg.flink.sink.committer.DeltaManifests}s in state for recovery, and commits
+ * them to the Iceberg table using the {@link CommonCommitter}
+ * <li>{@link WithPostCommitTopology} we could use for incremental compaction later. This is not
+ * implemented yet.
+ * </ul>
+ *
+ * The job graph looks like below:
+ *
+ * <pre>{@code
+ * Flink sink
+ * +-----------------------------------------------------------------------------------+
+ * | |
+ * +-------+ | +----------+ +-------------+ +---------------+ |
+ * | Map 1 | ==> | | writer 1 | | committer 1 | ---> | post commit 1 | |
+ * +-------+ | +----------+ +-------------+ +---------------+ |
+ * | \ / \ |
+ * | \ / \ |
+ * | \ / \ |
+ * +-------+ | +----------+ \ +-------------------+ / +-------------+ \ +---------------+ |
+ * | Map 2 | ==> | | writer 2 | --->| commit aggregator | | committer 2 | | post commit 2 | |
+ * +-------+ | +----------+ +-------------------+ +-------------+ +---------------+ |
+ * | Commit only on |
+ * | committer 1 |
+ * +-----------------------------------------------------------------------------------+
+ * }</pre>
+ */
+public class IcebergSink extends SinkBase
+ implements WithPreWriteTopology<RowData>,
+ WithPreCommitTopology<RowData, SinkV2Committable>,
+ WithPostCommitTopology<RowData, SinkV2Committable> {
+ private Table initTable;
+ private final TableLoader tableLoader;
+ private final Map<String, String> snapshotProperties;
+ private String uidPrefix;
+ private Committer<SinkV2Committable> sinkV2Committer;
+ private final String sinkId;
+ private CommonCommitter commonCommitter;
+
+ // FlinkWriteConf properties stored here for serialization
+ private boolean upsertMode;
+ private FileFormat dataFileFormat;
+ private long targetDataFileSize;
+
+ private IcebergSink(SinkBuilder builder) {
+ super(builder);
+
+ this.tableLoader = builder.tableLoader();
+ this.snapshotProperties = builder.snapshotProperties();
+
+ // We generate a random UUID every time when a sink is created.
+ // This is used to separate files generated by different sinks writing the same table.
+ // Also used to generate the aggregator operator name
+ this.sinkId = UUID.randomUUID().toString();
+ }
+
+ @Override
+ void init() {
+ super.init();
+ this.initTable = tableSupplier().get();
+ this.uidPrefix = builder().uidPrefix() == null ? "" : builder().uidPrefix();
+
+ // FlinkWriteConf properties needed to be set separately, so we do not have to serialize the
+ // full conf
+ this.upsertMode = flinkWriteConf().upsertMode();
+ this.dataFileFormat = flinkWriteConf().dataFileFormat();
+ this.targetDataFileSize = flinkWriteConf().targetDataFileSize();
+
+ this.commonCommitter =
+ new CommonCommitter(
+ tableLoader,
+ flinkWriteConf().branch(),
+ snapshotProperties,
+ flinkWriteConf().overwriteMode(),
+ flinkWriteConf().workerPoolSize(),
+ sinkId);
+ }
+
+ @Override
+ public DataStream<RowData> addPreWriteTopology(DataStream<RowData> inputDataStream) {
+ return distributeDataStream(inputDataStream);
+ }
+
+ @Override
+ public SinkV2Writer createWriter(InitContext context) {
+ RowDataTaskWriterFactory taskWriterFactory =
+ new RowDataTaskWriterFactory(
+ tableSupplier(),
+ flinkRowType(),
+ targetDataFileSize,
+ dataFileFormat,
+ writeProperties(),
+ equalityFieldIds(),
+ upsertMode);
+ IcebergStreamWriterMetrics metrics =
+ new IcebergStreamWriterMetrics(context.metricGroup(), initTable.name());
+ return new SinkV2Writer(
+ tableSupplier(),
+ taskWriterFactory,
+ metrics,
+ context.getSubtaskId(),
+ context.getAttemptNumber());
+ }
+
+ @Override
+ public DataStream<CommittableMessage<SinkV2Committable>> addPreCommitTopology(
+ DataStream<CommittableMessage<SinkV2Committable>> writeResults) {
+ TypeInformation<CommittableMessage<SinkV2Committable>> typeInformation =
+ CommittableMessageTypeInfo.of(this::getCommittableSerializer);
+ return writeResults
+ .global()
+ .transform(
+ prefixIfNotNull(uidPrefix, initTable.name() + "-" + sinkId + "-pre-commit-topology"),
+ typeInformation,
+ new SinkV2Aggregator(commonCommitter))
+ .uid(prefixIfNotNull(uidPrefix, "pre-commit-topology"))
+ .setParallelism(1)
+ .setMaxParallelism(1)
+ .global();
Review Comment:
These three could be replaced by `.forceNonParallel()`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: new sink base on the unified sink API - WIP [iceberg]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #8653:
URL: https://github.com/apache/iceberg/pull/8653#discussion_r1345697890
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/SinkBase.java:
##########
@@ -0,0 +1,326 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.sink;
+
+import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION;
+import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION_LEVEL;
+import static org.apache.iceberg.TableProperties.ORC_COMPRESSION;
+import static org.apache.iceberg.TableProperties.ORC_COMPRESSION_STRATEGY;
+import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION;
+import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_LEVEL;
+import static org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.DistributionMode;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SerializableTable;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.FlinkWriteConf;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.SerializableSupplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SinkBase implements Serializable {
+ private static final Logger LOG = LoggerFactory.getLogger(SinkBase.class);
+
+ private transient SinkBuilder builder;
+ private List<Integer> equalityFieldIds;
+ private RowType flinkRowType;
+ // Can not be serialized, so we have to be careful and serialize specific fields when needed
+ private transient FlinkWriteConf flinkWriteConf;
+ private Map<String, String> writeProperties;
+ private SerializableSupplier<Table> tableSupplier;
+
+ SinkBase(SinkBuilder builder) {
+ this.builder = builder;
+ }
+
+ SinkBuilder builder() {
+ return builder;
+ }
+
+ FlinkWriteConf flinkWriteConf() {
+ return flinkWriteConf;
+ }
+
+ Map<String, String> writeProperties() {
+ return writeProperties;
+ }
+
+ SerializableSupplier<Table> tableSupplier() {
+ return tableSupplier;
+ }
+
+ RowType flinkRowType() {
+ return flinkRowType;
+ }
+
+ List<Integer> equalityFieldIds() {
Review Comment:
Why not simply protected / package private fields and access them directly to reduce boilerplate?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org