This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5a8b59777 [flink] Generate tags when Flink Batch is completed
5a8b59777 is described below
commit 5a8b59777f0790e15d8932b7af33a03501b71fd8
Author: siyang.zeng <[email protected]>
AuthorDate: Thu Dec 7 19:46:10 2023 +0800
[flink] Generate tags when Flink Batch is completed
This closes #2469
---
docs/content/maintenance/manage-tags.md | 1 +
.../shortcodes/generated/core_configuration.html | 4 +-
.../main/java/org/apache/paimon/CoreOptions.java | 6 +-
.../java/org/apache/paimon/tag/TagPreview.java | 3 +-
.../org/apache/paimon/tag/TagTimeExtractor.java | 1 +
.../flink/sink/BatchWriteGeneratorTagOperator.java | 249 +++++++++++++++++++++
.../org/apache/paimon/flink/sink/FlinkSink.java | 8 +
.../sink/BatchWriteGeneratorTagOperatorTest.java | 102 +++++++++
8 files changed, 369 insertions(+), 5 deletions(-)
diff --git a/docs/content/maintenance/manage-tags.md
b/docs/content/maintenance/manage-tags.md
index 76afb7747..865e68b5e 100644
--- a/docs/content/maintenance/manage-tags.md
+++ b/docs/content/maintenance/manage-tags.md
@@ -42,6 +42,7 @@ Paimon supports automatic creation of tags in writing job.
You can set `'tag.automatic-creation'` to `process-time` or `watermark`:
- `process-time`: Create TAG based on the time of the machine.
- `watermark`: Create TAG based on the watermark of the Sink input.
+- `batch`: In a batch processing scenario, a tag is generated after the
current task is completed.
{{< hint info >}}
If you choose Watermark, you may need to specify the time zone of watermark,
if watermark is not in the
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 1c3c747f6..875286150 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -309,7 +309,7 @@ This config option does not affect the default filesystem
metastore.</td>
<td><h5>metastore.tag-to-partition.preview</h5></td>
<td style="word-wrap: break-word;">none</td>
<td><p>Enum</p></td>
- <td>Whether to preview tag of generated snapshots in metastore.
This allows the Hive engine to query specific tag before creation.<br /><br
/>Possible values:<ul><li>"none": No automatically created
tags.</li><li>"process-time": Based on the time of the machine, create TAG once
the processing time passes period time plus delay.</li><li>"watermark": Based
on the watermark of the input, create TAG once the watermark passes period time
plus delay.</li></ul></td>
+ <td>Whether to preview tag of generated snapshots in metastore.
This allows the Hive engine to query specific tag before creation.<br /><br
/>Possible values:<ul><li>"none": No automatically created
tags.</li><li>"process-time": Based on the time of the machine, create TAG once
the processing time passes period time plus delay.</li><li>"watermark": Based
on the watermark of the input, create TAG once the watermark passes period time
plus delay.</li><li>"batch": In the batch p [...]
</tr>
<tr>
<td><h5>num-levels</h5></td>
@@ -579,7 +579,7 @@ This config option does not affect the default filesystem
metastore.</td>
<td><h5>tag.automatic-creation</h5></td>
<td style="word-wrap: break-word;">none</td>
<td><p>Enum</p></td>
- <td>Whether to create tag automatically. And how to generate
tags.<br /><br />Possible values:<ul><li>"none": No automatically created
tags.</li><li>"process-time": Based on the time of the machine, create TAG once
the processing time passes period time plus delay.</li><li>"watermark": Based
on the watermark of the input, create TAG once the watermark passes period time
plus delay.</li></ul></td>
+ <td>Whether to create tag automatically. And how to generate
tags.<br /><br />Possible values:<ul><li>"none": No automatically created
tags.</li><li>"process-time": Based on the time of the machine, create TAG once
the processing time passes period time plus delay.</li><li>"watermark": Based
on the watermark of the input, create TAG once the watermark passes period time
plus delay.</li><li>"batch": In the batch processing scenario, the tag
corresponding to the current snapsho [...]
</tr>
<tr>
<td><h5>tag.callback.#.param</h5></td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 90f74ad2a..e3f410a87 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1966,8 +1966,10 @@ public class CoreOptions implements Serializable {
"Based on the time of the machine, create TAG once the
processing time passes period time plus delay."),
WATERMARK(
"watermark",
- "Based on the watermark of the input, create TAG once the
watermark passes period time plus delay.");
-
+ "Based on the watermark of the input, create TAG once the
watermark passes period time plus delay."),
+ BATCH(
+ "batch",
+ "In the batch processing scenario, the tag corresponding to
the current snapshot is generated after the task is completed.");
private final String value;
private final String description;
diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/TagPreview.java
b/paimon-core/src/main/java/org/apache/paimon/tag/TagPreview.java
index 671e8841d..cd1a3fda1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/TagPreview.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/TagPreview.java
@@ -47,7 +47,8 @@ public class TagPreview {
}
public static TagPreview create(CoreOptions options) {
- if (options.tagToPartitionPreview() !=
CoreOptions.TagCreationMode.NONE) {
+ if (options.tagToPartitionPreview() != CoreOptions.TagCreationMode.NONE
+ && options.tagToPartitionPreview() !=
CoreOptions.TagCreationMode.BATCH) {
return new TagPreview(options);
}
return null;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java
b/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java
index dadfde3a7..e017b2151 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExtractor.java
@@ -79,6 +79,7 @@ public interface TagTimeExtractor {
static TagTimeExtractor create(CoreOptions.TagCreationMode mode,
CoreOptions options) {
switch (mode) {
case NONE:
+ case BATCH:
return null;
case PROCESS_TIME:
return new ProcessTimeExtractor();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperator.java
new file mode 100644
index 000000000..dff75ff29
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperator.java
@@ -0,0 +1,249 @@
+/*
+ * 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.paimon.flink.sink;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.operation.TagDeletion;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
+import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.SetupableStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamTaskStateInitializer;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.SortedMap;
+
+/**
+ * Commit {@link Committable} for snapshot using the {@link
CommitterOperator}. When the task is
+ * completed, the corresponding tag is generated.
+ */
+public class BatchWriteGeneratorTagOperator<CommitT, GlobalCommitT>
+ implements OneInputStreamOperator<CommitT, CommitT>,
+ SetupableStreamOperator,
+ BoundedOneInput {
+
+ private static final String BATCH_WRITE_TAG_PREFIX = "batch-write-";
+
+ private static final long serialVersionUID = 1L;
+
+ private final CommitterOperator<CommitT, GlobalCommitT> commitOperator;
+
+ protected final FileStoreTable table;
+
+ public BatchWriteGeneratorTagOperator(
+ CommitterOperator<CommitT, GlobalCommitT> commitOperator,
FileStoreTable table) {
+ this.table = table;
+ this.commitOperator = commitOperator;
+ }
+
+ @Override
+ public void initializeState(StreamTaskStateInitializer
streamTaskStateManager)
+ throws Exception {
+ commitOperator.initializeState(streamTaskStateManager);
+ }
+
+ @Override
+ public OperatorSnapshotFutures snapshotState(
+ long checkpointId,
+ long timestamp,
+ CheckpointOptions checkpointOptions,
+ CheckpointStreamFactory storageLocation)
+ throws Exception {
+ return commitOperator.snapshotState(
+ checkpointId, timestamp, checkpointOptions, storageLocation);
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ commitOperator.notifyCheckpointComplete(checkpointId);
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) throws Exception {
+ commitOperator.notifyCheckpointAborted(checkpointId);
+ }
+
+ private void createTag() {
+ SnapshotManager snapshotManager = table.snapshotManager();
+ Snapshot snapshot = snapshotManager.latestSnapshot();
+ if (snapshot == null) {
+ return;
+ }
+ TagManager tagManager = table.tagManager();
+ TagDeletion tagDeletion = table.store().newTagDeletion();
+ Instant instant = Instant.ofEpochMilli(snapshot.timeMillis());
+ LocalDateTime localDateTime = LocalDateTime.ofInstant(instant,
ZoneId.systemDefault());
+ String tagName =
+ BATCH_WRITE_TAG_PREFIX
+ +
localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
+ try {
+ // If the tag already exists, delete the tag
+ if (tagManager.tagExists(tagName)) {
+ tagManager.deleteTag(tagName, tagDeletion, snapshotManager);
+ }
+ // Create a new tag
+ tagManager.createTag(snapshot, tagName);
+ // Expire the tag
+ expireTag();
+ } catch (Exception e) {
+ if (tagManager.tagExists(tagName)) {
+ tagManager.deleteTag(tagName, tagDeletion, snapshotManager);
+ }
+ }
+ }
+
+ private void expireTag() {
+ Integer tagNumRetainedMax = table.coreOptions().tagNumRetainedMax();
+ if (tagNumRetainedMax != null) {
+ SnapshotManager snapshotManager = table.snapshotManager();
+ if (snapshotManager.latestSnapshot() == null) {
+ return;
+ }
+ TagManager tagManager = table.tagManager();
+ TagDeletion tagDeletion = table.store().newTagDeletion();
+ SortedMap<Snapshot, String> tags = tagManager.tags();
+ if (tags.size() > tagNumRetainedMax) {
+ int toDelete = tags.size() - tagNumRetainedMax;
+ int i = 0;
+ for (String tag : tags.values()) {
+ tagManager.deleteTag(tag, tagDeletion, snapshotManager);
+ i++;
+ if (i == toDelete) {
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void open() throws Exception {
+ commitOperator.open();
+ }
+
+ @Override
+ public void processElement(StreamRecord<CommitT> element) throws Exception
{
+ commitOperator.processElement(element);
+ }
+
+ @Override
+ public void processWatermark(Watermark watermark) throws Exception {
+ commitOperator.processWatermark(watermark);
+ }
+
+ @Override
+ public void processWatermarkStatus(WatermarkStatus watermarkStatus) throws
Exception {
+ commitOperator.processWatermarkStatus(watermarkStatus);
+ }
+
+ @Override
+ public void processLatencyMarker(LatencyMarker latencyMarker) throws
Exception {
+ commitOperator.processLatencyMarker(latencyMarker);
+ }
+
+ @Override
+ public void finish() throws Exception {
+ createTag();
+ commitOperator.finish();
+ }
+
+ @Override
+ public void close() throws Exception {
+ commitOperator.close();
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ commitOperator.prepareSnapshotPreBarrier(checkpointId);
+ }
+
+ @Override
+ public void setKeyContextElement1(StreamRecord<?> record) throws Exception
{
+ commitOperator.setKeyContextElement1(record);
+ }
+
+ @Override
+ public void setKeyContextElement2(StreamRecord<?> record) throws Exception
{
+ commitOperator.setKeyContextElement2(record);
+ }
+
+ @Override
+ public OperatorMetricGroup getMetricGroup() {
+ return commitOperator.getMetricGroup();
+ }
+
+ @Override
+ public OperatorID getOperatorID() {
+ return commitOperator.getOperatorID();
+ }
+
+ @Override
+ public void setCurrentKey(Object key) {
+ commitOperator.setCurrentKey(key);
+ }
+
+ @Override
+ public Object getCurrentKey() {
+ return commitOperator.getCurrentKey();
+ }
+
+ @Override
+ public void setKeyContextElement(StreamRecord<CommitT> record) throws
Exception {
+ commitOperator.setKeyContextElement(record);
+ }
+
+ @Override
+ public void endInput() throws Exception {
+ commitOperator.endInput();
+ }
+
+ @Override
+ public void setup(StreamTask containingTask, StreamConfig config, Output
output) {
+ commitOperator.setup(containingTask, config, output);
+ }
+
+ @Override
+ public ChainingStrategy getChainingStrategy() {
+ return commitOperator.getChainingStrategy();
+ }
+
+ @Override
+ public void setChainingStrategy(ChainingStrategy strategy) {
+ commitOperator.setChainingStrategy(strategy);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index c9a6f3c05..480d41ae8 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.CoreOptions.ChangelogProducer;
+import org.apache.paimon.CoreOptions.TagCreationMode;
import org.apache.paimon.flink.utils.StreamExecutionEnvironmentUtils;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.options.MemorySize;
@@ -203,6 +204,13 @@ public abstract class FlinkSink<T> implements Serializable
{
table::tagManager,
() -> table.store().newTagDeletion());
}
+ if (conf.get(ExecutionOptions.RUNTIME_MODE) ==
RuntimeExecutionMode.BATCH
+ && table.coreOptions().tagCreationMode() ==
TagCreationMode.BATCH) {
+ committerOperator =
+ new BatchWriteGeneratorTagOperator<>(
+ (CommitterOperator<Committable,
ManifestCommittable>) committerOperator,
+ table);
+ }
SingleOutputStreamOperator<?> committed =
written.transform(
GLOBAL_COMMITTER_NAME + " : " + table.name(),
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperatorTest.java
new file mode 100644
index 000000000..894410dae
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/BatchWriteGeneratorTagOperatorTest.java
@@ -0,0 +1,102 @@
+/*
+ * 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.paimon.flink.sink;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.flink.VersionedSerializerWrapper;
+import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.manifest.ManifestCommittableSerializer;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
+import java.util.Objects;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link BatchWriteGeneratorTagOperator}. */
+public class BatchWriteGeneratorTagOperatorTest extends CommitterOperatorTest {
+
+ @Test
+ public void testBatchWriteGeneratorTag() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ // set tag.automatic-creation = batch
+ HashMap<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put("tag.automatic-creation", "batch");
+ dynamicOptions.put("tag.num-retained-max", "2");
+ table = table.copy(dynamicOptions);
+
+ StreamTableWrite write =
+
table.newStreamWriteBuilder().withCommitUser(initialCommitUser).newWrite();
+
+ OneInputStreamOperator<Committable, Committable> committerOperator =
+ createCommitterOperator(
+ table,
+ initialCommitUser,
+ new RestoreAndFailCommittableStateManager<>(
+ () ->
+ new VersionedSerializerWrapper<>(
+ new
ManifestCommittableSerializer())));
+
+ TableCommitImpl tableCommit = table.newCommit(initialCommitUser);
+
+ write.write(GenericRow.of(1, 10L));
+ tableCommit.commit(write.prepareCommit(false, 1));
+
+ SnapshotManager snapshotManager =
table.newSnapshotReader().snapshotManager();
+ TagManager tagManager = table.tagManager();
+
+ // Generate tag name
+ String prefix = "batch-write-";
+ Instant instant =
+ Instant.ofEpochMilli(
+
Objects.requireNonNull(snapshotManager.latestSnapshot()).timeMillis());
+ LocalDateTime localDateTime = LocalDateTime.ofInstant(instant,
ZoneId.systemDefault());
+ String tagName = prefix +
localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
+
+ // No tag is generated before the finish method
+ assertThat(table.tagManager().tagCount()).isEqualTo(0);
+ committerOperator.finish();
+ // After the finish method, a tag is generated
+ assertThat(table.tagManager().tagCount()).isEqualTo(1);
+ // The tag is consistent with the latest snapshot
+
assertThat(tagManager.taggedSnapshot(tagName)).isEqualTo(snapshotManager.latestSnapshot());
+ }
+
+ @Override
+ protected OneInputStreamOperator<Committable, Committable>
createCommitterOperator(
+ FileStoreTable table,
+ String commitUser,
+ CommittableStateManager<ManifestCommittable>
committableStateManager) {
+ return new BatchWriteGeneratorTagOperator<>(
+ (CommitterOperator<Committable, ManifestCommittable>)
+ super.createCommitterOperator(table, commitUser,
committableStateManager),
+ table);
+ }
+}
