You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by mx...@apache.org on 2019/11/05 12:16:00 UTC
[beam] branch master updated: [BEAM-8549] Do not use keyed operator
state for checkpoint buffering
This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new c81b5f4 [BEAM-8549] Do not use keyed operator state for checkpoint buffering
new 9f56639 Merge pull request #9980 from mxm/BEAM-8549
c81b5f4 is described below
commit c81b5f4ab02c2a7549a319dc9cda133b2785be9a
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Sun Nov 3 21:09:20 2019 +0100
[BEAM-8549] Do not use keyed operator state for checkpoint buffering
The current buffer logic for items emitted during checkpointing is faulty in the
sense that the buffer is partitioned on the output keys of the operator. The key
may be changed or even be dropped. Thus, the original key partitioning will not
be maintained which will cause checkpointing to fail.
An alternative solution would be BEAM-6733 / #9652, but this change keeps the
current buffering logic in place. The output buffer may now always be
redistributed round-robin upon restoring from a checkpoint. Note that this is
fine because no assumption can be made about the distribution of output elements
of a DoFn operation.
---
.../wrappers/streaming/DoFnOperator.java | 36 ++++---------------
.../wrappers/streaming/DoFnOperatorTest.java | 41 ++++++++++++----------
2 files changed, 29 insertions(+), 48 deletions(-)
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
index b4ce64f..34250a5 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
@@ -388,11 +388,7 @@ public class DoFnOperator<InputT, OutputT> extends AbstractStreamOperator<Window
outputManager =
outputManagerFactory.create(
- output,
- getLockToAcquireForStateAccessDuringBundles(),
- getOperatorStateBackend(),
- getKeyedStateBackend(),
- keySelector);
+ output, getLockToAcquireForStateAccessDuringBundles(), getOperatorStateBackend());
}
/**
@@ -828,9 +824,7 @@ public class DoFnOperator<InputT, OutputT> extends AbstractStreamOperator<Window
BufferedOutputManager<OutputT> create(
Output<StreamRecord<WindowedValue<OutputT>>> output,
Lock bufferLock,
- @Nullable OperatorStateBackend operatorStateBackend,
- @Nullable KeyedStateBackend keyedStateBackend,
- @Nullable KeySelector keySelector)
+ OperatorStateBackend operatorStateBackend)
throws Exception;
}
@@ -1018,35 +1012,19 @@ public class DoFnOperator<InputT, OutputT> extends AbstractStreamOperator<Window
public BufferedOutputManager<OutputT> create(
Output<StreamRecord<WindowedValue<OutputT>>> output,
Lock bufferLock,
- OperatorStateBackend operatorStateBackend,
- @Nullable KeyedStateBackend keyedStateBackend,
- @Nullable KeySelector keySelector)
+ OperatorStateBackend operatorStateBackend)
throws Exception {
Preconditions.checkNotNull(output);
Preconditions.checkNotNull(bufferLock);
Preconditions.checkNotNull(operatorStateBackend);
- Preconditions.checkState(
- (keyedStateBackend == null) == (keySelector == null),
- "Either both KeyedStatebackend and Keyselector are provided or none.");
TaggedKvCoder taggedKvCoder = buildTaggedKvCoder();
ListStateDescriptor<KV<Integer, WindowedValue<?>>> taggedOutputPushbackStateDescriptor =
new ListStateDescriptor<>("bundle-buffer-tag", new CoderTypeSerializer<>(taggedKvCoder));
-
- final PushedBackElementsHandler<KV<Integer, WindowedValue<?>>> pushedBackElementsHandler;
- if (keyedStateBackend != null) {
- // build a key selector for the tagged output
- KeySelector<KV<Integer, WindowedValue<?>>, ?> taggedValueKeySelector =
- (KeySelector<KV<Integer, WindowedValue<?>>, Object>)
- value -> keySelector.getKey(value.getValue());
- pushedBackElementsHandler =
- KeyedPushedBackElementsHandler.create(
- taggedValueKeySelector, keyedStateBackend, taggedOutputPushbackStateDescriptor);
- } else {
- ListState<KV<Integer, WindowedValue<?>>> listState =
- operatorStateBackend.getListState(taggedOutputPushbackStateDescriptor);
- pushedBackElementsHandler = NonKeyedPushedBackElementsHandler.create(listState);
- }
+ ListState<KV<Integer, WindowedValue<?>>> listStateBuffer =
+ operatorStateBackend.getListState(taggedOutputPushbackStateDescriptor);
+ PushedBackElementsHandler<KV<Integer, WindowedValue<?>>> pushedBackElementsHandler =
+ NonKeyedPushedBackElementsHandler.create(listStateBuffer);
return new BufferedOutputManager<>(
output, mainTag, tagsToOutputTags, tagsToIds, bufferLock, pushedBackElementsHandler);
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java
index 6e96b3b..c5eca1e 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperatorTest.java
@@ -1313,20 +1313,25 @@ public class DoFnOperatorTest {
options.setMaxBundleSize(2L);
options.setMaxBundleTimeMills(10L);
- IdentityDoFn<KV<String, String>> doFn =
- new IdentityDoFn<KV<String, String>>() {
+ DoFn<KV<String, String>, String> doFn =
+ new DoFn<KV<String, String>, String>() {
+ @ProcessElement
+ public void processElement(ProcessContext ctx) {
+ // Change output type of element to test that we do not depend on the input keying
+ ctx.output(ctx.element().getValue());
+ }
+
@FinishBundle
public void finishBundle(FinishBundleContext context) {
context.output(
- KV.of("key2", "finishBundle"),
- BoundedWindow.TIMESTAMP_MIN_VALUE,
- GlobalWindow.INSTANCE);
+ "finishBundle", BoundedWindow.TIMESTAMP_MIN_VALUE, GlobalWindow.INSTANCE);
}
};
- DoFnOperator.MultiOutputOutputManagerFactory<KV<String, String>> outputManagerFactory =
+ DoFnOperator.MultiOutputOutputManagerFactory<String> outputManagerFactory =
new DoFnOperator.MultiOutputOutputManagerFactory(
- outputTag, WindowedValue.getFullCoder(kvCoder, GlobalWindow.Coder.INSTANCE));
+ outputTag,
+ WindowedValue.getFullCoder(kvCoder.getValueCoder(), GlobalWindow.Coder.INSTANCE));
DoFnOperator<KV<String, String>, KV<String, String>> doFnOperator =
new DoFnOperator(
@@ -1347,8 +1352,7 @@ public class DoFnOperatorTest {
DoFnSchemaInformation.create(),
Collections.emptyMap());
- OneInputStreamOperatorTestHarness<
- WindowedValue<KV<String, String>>, WindowedValue<KV<String, String>>>
+ OneInputStreamOperatorTestHarness<WindowedValue<KV<String, String>>, WindowedValue<String>>
testHarness =
new KeyedOneInputStreamOperatorTestHarness(
doFnOperator, keySelector, keySelector.getProducedType());
@@ -1365,10 +1369,10 @@ public class DoFnOperatorTest {
assertThat(
stripStreamRecordFromWindowedValue(testHarness.getOutput()),
contains(
- WindowedValue.valueInGlobalWindow(KV.of("key", "a")),
- WindowedValue.valueInGlobalWindow(KV.of("key", "b")),
- WindowedValue.valueInGlobalWindow(KV.of("key2", "finishBundle")),
- WindowedValue.valueInGlobalWindow(KV.of("key", "c"))));
+ WindowedValue.valueInGlobalWindow("a"),
+ WindowedValue.valueInGlobalWindow("b"),
+ WindowedValue.valueInGlobalWindow("finishBundle"),
+ WindowedValue.valueInGlobalWindow("c")));
// Take a snapshot
OperatorSubtaskState snapshot = testHarness.snapshot(0, 0);
@@ -1376,12 +1380,11 @@ public class DoFnOperatorTest {
// Finish bundle element will be buffered as part of finishing a bundle in snapshot()
PushedBackElementsHandler<KV<Integer, WindowedValue<?>>> pushedBackElementsHandler =
doFnOperator.outputManager.pushedBackElementsHandler;
- assertThat(pushedBackElementsHandler, instanceOf(KeyedPushedBackElementsHandler.class));
+ assertThat(pushedBackElementsHandler, instanceOf(NonKeyedPushedBackElementsHandler.class));
List<KV<Integer, WindowedValue<?>>> bufferedElements =
pushedBackElementsHandler.getElements().collect(Collectors.toList());
assertThat(
- bufferedElements,
- contains(KV.of(0, WindowedValue.valueInGlobalWindow(KV.of("key2", "finishBundle")))));
+ bufferedElements, contains(KV.of(0, WindowedValue.valueInGlobalWindow("finishBundle"))));
testHarness.close();
@@ -1424,9 +1427,9 @@ public class DoFnOperatorTest {
stripStreamRecordFromWindowedValue(testHarness.getOutput()),
contains(
// The first finishBundle is restored from the checkpoint
- WindowedValue.valueInGlobalWindow(KV.of("key2", "finishBundle")),
- WindowedValue.valueInGlobalWindow(KV.of("key", "d")),
- WindowedValue.valueInGlobalWindow(KV.of("key2", "finishBundle"))));
+ WindowedValue.valueInGlobalWindow("finishBundle"),
+ WindowedValue.valueInGlobalWindow("d"),
+ WindowedValue.valueInGlobalWindow("finishBundle")));
testHarness.close();
}