diff --git a/CHANGES.md b/CHANGES.md index f98790bc2e19..3e616f1f3651 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -68,6 +68,7 @@ ## New Features / Improvements +* (Java) Flink runner: the batch-only `--sortedBatchCombine` option is enabled by default (disable it with `--sortedBatchCombine=false`). GroupByKey and Combine in the global window with the default trigger then finish with a dedicated operator that relies on Flink sorting its input by key, instead of keyed state and timers ([#40607](https://github.com/apache/beam/issues/40607)). * (Go) Added `wait.On`, which delays each input window until the corresponding windows in its signal PCollections have closed ([#39909](https://github.com/apache/beam/issues/39909)). * (Python) Expanded the SDK worker heap dump (`--experiments=enable_heap_dump`) with process RSS, CPython allocator/GC stats, and glibc `mallinfo2` native-heap/fragmentation stats to help distinguish native-heap from Python-object memory growth ([#39244](https://github.com/apache/beam/issues/39244)). * The `disableCounterMetrics`, `disableStringSetMetrics` and `disableBoundedTrieMetrics` experiments are now honored by the Python SDK, as they already were in Java (Python) ([#38746](https://github.com/apache/beam/issues/38746)). diff --git a/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java b/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java index d3c5fefd9cf7..b5292a734e8b 100644 --- a/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java +++ b/runners/flink/2.0/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java @@ -350,6 +350,16 @@ public Long create(PipelineOptions options) { void setFasterCopy(Boolean fasterCopy); + @Description( + "Batch only. Use a dedicated operator for the final step of GroupByKey and Combine in the " + + "global window with the default trigger. It relies on Flink sorting the input by key " + + "and does not use keyed state or timers. Has no effect in streaming mode. " + + "Enabled by default; set to false to disable.") + @Default.Boolean(true) + Boolean getSortedBatchCombine(); + + void setSortedBatchCombine(Boolean sortedBatchCombine); + @Description( "Directory containing Flink YAML configuration files. " + "These properties will be set to all jobs submitted to Flink and take precedence " diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java index a6729356bdcb..b64737281940 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java @@ -367,6 +367,16 @@ public Long create(PipelineOptions options) { void setFasterCopy(Boolean fasterCopy); + @Description( + "Batch only. Use a dedicated operator for the final step of GroupByKey and Combine in the " + + "global window with the default trigger. It relies on Flink sorting the input by key " + + "and does not use keyed state or timers. Has no effect in streaming mode. " + + "Enabled by default; set to false to disable.") + @Default.Boolean(true) + Boolean getSortedBatchCombine(); + + void setSortedBatchCombine(Boolean sortedBatchCombine); + @Description( "Directory containing Flink YAML configuration files. " + "These properties will be set to all jobs submitted to Flink and take precedence " diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslators.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslators.java index 60448db365df..9de9c6310832 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslators.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslators.java @@ -22,6 +22,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.function.Supplier; import org.apache.beam.runners.core.KeyedWorkItem; import org.apache.beam.runners.core.SystemReduceFn; import org.apache.beam.runners.core.construction.SerializablePipelineOptions; @@ -31,6 +32,7 @@ import org.apache.beam.runners.flink.translation.wrappers.streaming.KvToFlinkKeyKeySelector; import org.apache.beam.runners.flink.translation.wrappers.streaming.PartialReduceBundleOperator; import org.apache.beam.runners.flink.translation.wrappers.streaming.SingletonKeyedWorkItemCoder; +import org.apache.beam.runners.flink.translation.wrappers.streaming.SortedBatchCombineOperator; import org.apache.beam.runners.flink.translation.wrappers.streaming.WindowDoFnOperator; import org.apache.beam.runners.flink.translation.wrappers.streaming.WorkItemKeySelector; import org.apache.beam.sdk.coders.CannotProvideCoderException; @@ -38,12 +40,15 @@ import org.apache.beam.sdk.coders.CoderRegistry; import org.apache.beam.sdk.coders.IterableCoder; import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.Combine; import org.apache.beam.sdk.transforms.CombineFnBase; import org.apache.beam.sdk.transforms.CombineWithContext; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.join.RawUnionValue; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.transforms.windowing.DefaultTrigger; +import org.apache.beam.sdk.transforms.windowing.GlobalWindows; import org.apache.beam.sdk.util.AppliedCombineFn; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; @@ -61,6 +66,7 @@ import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; import org.apache.flink.streaming.api.transformations.TwoInputTransformation; import org.apache.flink.util.Collector; @@ -279,7 +285,8 @@ SingleOutputStreamOperator>> getBatchCombinePerKeyO List> sideInputs, Coder>> windowedAccumCoder, CombineFnBase.GlobalCombineFn combineFn, - WindowDoFnOperator finalDoFnOperator, + OneInputStreamOperator>, WindowedValue>> + finalDoFnOperator, TypeInformation>> outputTypeInfo) { String fullName = FlinkStreamingTransformTranslators.getCurrentTransformName(context); @@ -348,7 +355,7 @@ SingleOutputStreamOperator>> getBatchCombinePerKeyO partialyCombinedStream.keyBy(accumKeySelector), transformSideInputs.f1, fullName, - finalDoFnOperator, + (WindowDoFnOperator) finalDoFnOperator, outputTypeInfo); } } @@ -400,9 +407,23 @@ SingleOutputStreamOperator>>> batchGroupByK serializablePipelineOptions); // final aggregation - WindowDoFnOperator, Iterable>> finalDoFnOperator = - getWindowedAccumulateDoFnOperator( - context, transform, accumKvCoder, outputCoder, sideInputTagMapping, sideInputs); + OneInputStreamOperator< + WindowedValue>>, + WindowedValue>>>> + finalDoFnOperator = + getFinalAggregationOperator( + context, + input, + sideInputs, + new SortedBatchCombineOperator.BufferingCombiner<>(), + () -> + getWindowedAccumulateDoFnOperator( + context, + transform, + accumKvCoder, + outputCoder, + sideInputTagMapping, + sideInputs)); return getBatchCombinePerKeyOperator( context, @@ -441,6 +462,47 @@ SingleOutputStreamOperator>>> batchGroupByK context, transform, accumKvCoder, outputCoder, reduceFn, sideInputTagMapping, sideInputs); } + /** + * The operator for the final step of a batch GroupByKey or Combine: a {@link + * SortedBatchCombineOperator} using {@code sortedCombiner} when {@link #useSortedBatchCombine} + * allows it, otherwise the {@link WindowDoFnOperator} from {@code windowDoFnOperator}. + */ + private static + OneInputStreamOperator>, WindowedValue>> + getFinalAggregationOperator( + FlinkStreamingTranslationContext context, + PCollection input, + List> sideInputs, + SortedBatchCombineOperator.KeyCombiner sortedCombiner, + Supplier> windowDoFnOperator) { + WindowingStrategy windowingStrategy = input.getWindowingStrategy(); + if (useSortedBatchCombine(context.getPipelineOptions(), windowingStrategy, sideInputs)) { + return new SortedBatchCombineOperator<>( + FlinkStreamingTransformTranslators.getCurrentTransformName(context), + sortedCombiner, + windowingStrategy.getTimestampCombiner(), + new SerializablePipelineOptions(context.getPipelineOptions())); + } else { + return windowDoFnOperator.get(); + } + } + + /** + * Whether the final step of a batch GroupByKey or Combine can use a {@link + * SortedBatchCombineOperator} instead of a {@link WindowDoFnOperator}. + */ + static boolean useSortedBatchCombine( + PipelineOptions pipelineOptions, + WindowingStrategy windowingStrategy, + List> sideInputs) { + FlinkPipelineOptions options = pipelineOptions.as(FlinkPipelineOptions.class); + return !options.isStreaming() + && options.getSortedBatchCombine() + && sideInputs.isEmpty() + && windowingStrategy.getWindowFn() instanceof GlobalWindows + && windowingStrategy.getTrigger() instanceof DefaultTrigger; + } + public static SingleOutputStreamOperator>> batchCombinePerKey( FlinkStreamingTranslationContext context, @@ -476,15 +538,22 @@ SingleOutputStreamOperator>> batchCombinePerKey( } // final aggregation from AccumT to OutputT - WindowDoFnOperator finalDoFnOperator = - getWindowedAggregateDoFnOperator( - context, - transform, - accumKvCoder, - outputCoder, - toFinalFlinkCombineFn(combineFn, inputKvCoder.getValueCoder()), - sideInputTagMapping, - sideInputs); + OneInputStreamOperator>, WindowedValue>> + finalDoFnOperator = + getFinalAggregationOperator( + context, + input, + sideInputs, + new SortedBatchCombineOperator.MergingCombiner<>(combineFn), + () -> + getWindowedAggregateDoFnOperator( + context, + transform, + accumKvCoder, + outputCoder, + toFinalFlinkCombineFn(combineFn, inputKvCoder.getValueCoder()), + sideInputTagMapping, + sideInputs)); return getBatchCombinePerKeyOperator( context, diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerBase.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerBase.java index e54d5575a479..1b3347eacf0c 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerBase.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerBase.java @@ -89,7 +89,7 @@ public void updateMetrics(String stepName, List monit * Update Flink's internal metrics ({@link this#flinkCounterCache}) with the latest metrics for a * given step. */ - void updateMetrics(String stepName) { + public void updateMetrics(String stepName) { MetricResults metricResults = asAttemptedOnlyMetricResults(metricsContainers); MetricQueryResults metricQueryResults = metricResults.queryMetrics(MetricsFilter.builder().addStep(stepName).build()); diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SortedBatchCombineOperator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SortedBatchCombineOperator.java new file mode 100644 index 000000000000..85a135f2c136 --- /dev/null +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SortedBatchCombineOperator.java @@ -0,0 +1,275 @@ +/* + * 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.beam.runners.flink.translation.wrappers.streaming; + +import java.io.Closeable; +import java.io.IOException; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.apache.beam.runners.core.GlobalCombineFnRunner; +import org.apache.beam.runners.core.GlobalCombineFnRunners; +import org.apache.beam.runners.core.NullSideInputReader; +import org.apache.beam.runners.core.SideInputReader; +import org.apache.beam.runners.core.construction.SerializablePipelineOptions; +import org.apache.beam.runners.flink.FlinkPipelineOptions; +import org.apache.beam.runners.flink.metrics.FlinkMetricContainer; +import org.apache.beam.sdk.metrics.MetricsEnvironment; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.transforms.CombineFnBase.GlobalCombineFn; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.transforms.windowing.GlobalWindow; +import org.apache.beam.sdk.transforms.windowing.PaneInfo; +import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.WindowedValue; +import org.apache.beam.sdk.values.WindowedValues; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.BoundedOneInput; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Instant; + +/** + * Final step of a batch GroupByKey or Combine in the global window with the default trigger. + * + *

In batch execution with sorted inputs, Flink feeds a keyed operator all the records of a key + * before moving to the next key. With the global window and the default trigger, each key produces + * exactly one on-time pane once all its records are seen. This operator therefore combines the + * records of the current key in plain fields, and emits the result when the key changes or the + * input ends. Unlike {@link WindowDoFnOperator}, it does not use keyed state, timers or watermark + * holds. + * + *

It must only be used on a keyed stream in batch execution, where Flink sorts the input by key. + */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) +}) +public class SortedBatchCombineOperator + extends AbstractStreamOperator>> + implements OneInputStreamOperator>, WindowedValue>>, + BoundedOneInput { + + private static final long serialVersionUID = 1L; + + /** Combines the accumulators of one key. */ + public interface KeyCombiner extends Serializable { + void add(AccumT accumulator, PipelineOptions options); + + /** Returns the output for the accumulators added since the last call, and resets. */ + OutputT finish(PipelineOptions options); + } + + /** Merges accumulators with a {@link GlobalCombineFn} and extracts its output. */ + public static class MergingCombiner implements KeyCombiner { + private static final long serialVersionUID = 1L; + + /** Number of accumulators merged at once, bounding the memory used per key. */ + static final int MERGE_BATCH_SIZE = 32; + + private final GlobalCombineFn combineFn; + private transient @Nullable GlobalCombineFnRunner runner; + private transient @Nullable List pending; + + public MergingCombiner(GlobalCombineFn combineFn) { + this.combineFn = combineFn; + } + + private GlobalCombineFnRunner runner() { + if (runner == null) { + runner = GlobalCombineFnRunners.create(combineFn); + } + return runner; + } + + private List pending() { + if (pending == null) { + pending = new ArrayList<>(MERGE_BATCH_SIZE); + } + return pending; + } + + @Override + public void add(AccumT accumulator, PipelineOptions options) { + List accumulators = pending(); + accumulators.add(accumulator); + if (accumulators.size() >= MERGE_BATCH_SIZE) { + AccumT merged = merge(accumulators, options); + accumulators.clear(); + accumulators.add(merged); + } + } + + @Override + public OutputT finish(PipelineOptions options) { + List accumulators = pending(); + AccumT merged = merge(accumulators, options); + accumulators.clear(); + return runner().extractOutput(merged, options, SIDE_INPUTS, WINDOWS); + } + + private AccumT merge(List accumulators, PipelineOptions options) { + return accumulators.size() == 1 + ? accumulators.get(0) + : runner().mergeAccumulators(accumulators, options, SIDE_INPUTS, WINDOWS); + } + } + + /** Collects the accumulators of a key, like {@code SystemReduceFn.buffering}. */ + public static class BufferingCombiner implements KeyCombiner> { + private static final long serialVersionUID = 1L; + + private transient @Nullable List pending; + + @Override + public void add(AccumT accumulator, PipelineOptions options) { + if (pending == null) { + pending = new ArrayList<>(); + } + pending.add(accumulator); + } + + @Override + public Iterable finish(PipelineOptions options) { + List result = pending == null ? new ArrayList<>() : pending; + pending = null; + return result; + } + } + + private static final SideInputReader SIDE_INPUTS = NullSideInputReader.empty(); + + private static final List WINDOWS = + Collections.singletonList(GlobalWindow.INSTANCE); + + private final String stepName; + private final KeyCombiner combiner; + private final TimestampCombiner timestampCombiner; + private final SerializablePipelineOptions serializedOptions; + + private transient @Nullable PipelineOptions options; + transient @Nullable FlinkMetricContainer metricContainer; + + // The key whose records are being combined. + private transient boolean hasCurrentKey; + private transient @Nullable Object currentFlinkKey; + private transient @Nullable K currentKey; + private transient @Nullable Instant outputTimestamp; + + public SortedBatchCombineOperator( + String stepName, + KeyCombiner combiner, + TimestampCombiner timestampCombiner, + SerializablePipelineOptions serializedOptions) { + this.stepName = stepName; + this.combiner = combiner; + this.timestampCombiner = timestampCombiner; + this.serializedOptions = serializedOptions; + } + + @Override + public void open() throws Exception { + super.open(); + options = serializedOptions.get(); + if (!options.as(FlinkPipelineOptions.class).getDisableMetrics()) { + metricContainer = new FlinkMetricContainer(getRuntimeContext()); + } + } + + @Override + public void processElement(StreamRecord>> record) throws Exception { + WindowedValue> element = record.getValue(); + // Set by Flink from the key selector of the keyed stream before each record. + Object flinkKey = getCurrentKey(); + if (hasCurrentKey && !flinkKey.equals(currentFlinkKey)) { + emitCurrentKey(); + } + KV kv = element.getValue(); + Instant timestamp = timestampCombiner.assign(GlobalWindow.INSTANCE, element.getTimestamp()); + if (!hasCurrentKey) { + hasCurrentKey = true; + currentFlinkKey = flinkKey; + currentKey = kv.getKey(); + outputTimestamp = timestamp; + } else { + outputTimestamp = timestampCombiner.combine(outputTimestamp, timestamp); + } + try (Closeable ignored = scopedMetrics()) { + combiner.add(kv.getValue(), options); + } + } + + @Override + public void processWatermark(Watermark mark) throws Exception { + // The output of the current key may be behind any watermark, so watermarks are held back until + // the end of the input. With sorted inputs, Flink only sends the final watermark, after the + // last + // record but before endInput(), so the last key must be emitted before forwarding it. + if (mark.getTimestamp() == Watermark.MAX_WATERMARK.getTimestamp()) { + if (hasCurrentKey) { + emitCurrentKey(); + } + super.processWatermark(mark); + } + } + + @Override + public void endInput() throws Exception { + if (hasCurrentKey) { + emitCurrentKey(); + } + if (metricContainer != null) { + metricContainer.updateMetrics(stepName); + } + } + + @Override + public void finish() throws Exception { + if (metricContainer != null) { + metricContainer.registerMetricsForPipelineResult(); + } + super.finish(); + } + + private void emitCurrentKey() throws IOException { + OutputT result; + try (Closeable ignored = scopedMetrics()) { + result = combiner.finish(options); + } + output.collect( + new StreamRecord<>( + WindowedValues.of( + KV.of(currentKey, result), + outputTimestamp, + GlobalWindow.INSTANCE, + PaneInfo.ON_TIME_AND_ONLY_FIRING))); + hasCurrentKey = false; + currentFlinkKey = null; + currentKey = null; + outputTimestamp = null; + } + + private Closeable scopedMetrics() { + return metricContainer == null + ? () -> {} + : MetricsEnvironment.scopedMetricsContainer(metricContainer.getMetricsContainer(stepName)); + } +} diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslatorsTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslatorsTest.java new file mode 100644 index 000000000000..236b240c16db --- /dev/null +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkStreamingAggregationsTranslatorsTest.java @@ -0,0 +1,89 @@ +/* + * 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.beam.runners.flink; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.Is.is; + +import java.util.Collections; +import java.util.List; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.View; +import org.apache.beam.sdk.transforms.windowing.AfterPane; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.Repeatedly; +import org.apache.beam.sdk.values.PCollectionView; +import org.apache.beam.sdk.values.WindowingStrategy; +import org.joda.time.Duration; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests {@link FlinkStreamingAggregationsTranslators}. */ +@RunWith(JUnit4.class) +public class FlinkStreamingAggregationsTranslatorsTest { + + private static FlinkPipelineOptions options(boolean streaming, boolean sortedBatchCombine) { + FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); + options.setStreaming(streaming); + options.setSortedBatchCombine(sortedBatchCombine); + return options; + } + + private static boolean eligible( + FlinkPipelineOptions options, + WindowingStrategy windowingStrategy, + List> sideInputs) { + return FlinkStreamingAggregationsTranslators.useSortedBatchCombine( + options, windowingStrategy, sideInputs); + } + + @Test + public void usesSortedBatchCombineOnlyWhenEligible() { + WindowingStrategy global = WindowingStrategy.globalDefault(); + List> none = Collections.emptyList(); + + assertThat(eligible(options(false, true), global, none), is(true)); + + assertThat("option off", eligible(options(false, false), global, none), is(false)); + assertThat("streaming", eligible(options(true, true), global, none), is(false)); + assertThat( + "fixed windows", + eligible( + options(false, true), + WindowingStrategy.of(FixedWindows.of(Duration.standardMinutes(1))), + none), + is(false)); + assertThat( + "custom trigger", + eligible( + options(false, true), + global.withTrigger(Repeatedly.forever(AfterPane.elementCountAtLeast(1))), + none), + is(false)); + + TestPipeline pipeline = TestPipeline.create(); + PCollectionView view = pipeline.apply(Create.of(1)).apply(View.asSingleton()); + assertThat( + "side inputs", + eligible(options(false, true), global, Collections.singletonList(view)), + is(false)); + } +} diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SortedBatchCombineOperatorTest.java b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SortedBatchCombineOperatorTest.java new file mode 100644 index 000000000000..8843fdc09df5 --- /dev/null +++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SortedBatchCombineOperatorTest.java @@ -0,0 +1,379 @@ +/* + * 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.beam.runners.flink.translation.wrappers.streaming; + +import static java.util.Collections.emptyList; +import static java.util.Collections.emptyMap; +import static org.apache.beam.runners.flink.translation.wrappers.streaming.StreamRecordStripper.stripStreamRecordFromWindowedValue; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.hamcrest.Matchers.emptyIterable; +import static org.hamcrest.core.Is.is; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import org.apache.beam.runners.core.KeyedWorkItem; +import org.apache.beam.runners.core.SystemReduceFn; +import org.apache.beam.runners.core.construction.SerializablePipelineOptions; +import org.apache.beam.runners.core.metrics.MetricsContainerImpl; +import org.apache.beam.runners.flink.FlinkPipelineOptions; +import org.apache.beam.runners.flink.adapter.FlinkKey; +import org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.MultiOutputOutputManagerFactory; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.CoderRegistry; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.VarLongCoder; +import org.apache.beam.sdk.metrics.MetricName; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.transforms.Combine; +import org.apache.beam.sdk.transforms.CombineWithContext; +import org.apache.beam.sdk.transforms.Sum; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.transforms.windowing.GlobalWindow; +import org.apache.beam.sdk.transforms.windowing.PaneInfo; +import org.apache.beam.sdk.transforms.windowing.TimestampCombiner; +import org.apache.beam.sdk.util.AppliedCombineFn; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.TupleTag; +import org.apache.beam.sdk.values.WindowedValue; +import org.apache.beam.sdk.values.WindowedValues; +import org.apache.beam.sdk.values.WindowedValues.FullWindowedValueCoder; +import org.apache.beam.sdk.values.WindowingStrategy; +import org.apache.flink.api.java.typeutils.ValueTypeInfo; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness; +import org.joda.time.Instant; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests {@link SortedBatchCombineOperator}. */ +@RunWith(JUnit4.class) +@SuppressWarnings({ + "rawtypes", // TODO(https://github.com/apache/beam/issues/20447) + "unchecked" +}) +public class SortedBatchCombineOperatorTest { + + private static final String STEP = "stepName"; + + /** Input grouped by key, as Flink provides it in batch: (key, value, timestamp). */ + private static List sortedInput() { + List input = new ArrayList<>(); + input.add(new long[] {1, 10, 5}); + // More inputs than MergingCombiner.MERGE_BATCH_SIZE. + for (int i = 0; i < 40; i++) { + input.add(new long[] {2, i, 100 - i}); + } + input.add(new long[] {3, 1, 7}); + input.add(new long[] {3, 2, 3}); + input.add(new long[] {3, 3, 9}); + return input; + } + + private static FlinkPipelineOptions options() { + FlinkPipelineOptions options = FlinkPipelineOptions.defaults(); + options.setStreaming(false); + return options; + } + + private static + KeyedOneInputStreamOperatorTestHarness< + FlinkKey, WindowedValue>, WindowedValue>> + harness( + OneInputStreamOperator< + WindowedValue>, WindowedValue>> + operator) + throws Exception { + return new KeyedOneInputStreamOperatorTestHarness<>( + operator, + new KvToFlinkKeyKeySelector<>(VarLongCoder.of()), + ValueTypeInfo.of(FlinkKey.class)); + } + + private static WindowedValue> element(long key, T value, long timestamp) { + return WindowedValues.of( + KV.of(key, value), new Instant(timestamp), GlobalWindow.INSTANCE, PaneInfo.NO_FIRING); + } + + /** The operator used today for the final step: WindowDoFnOperator with a combining ReduceFn. */ + private static WindowDoFnOperator windowDoFnOperator( + TimestampCombiner timestampCombiner) { + WindowingStrategy windowingStrategy = + (WindowingStrategy) + WindowingStrategy.globalDefault().withTimestampCombiner(timestampCombiner); + TupleTag> outputTag = new TupleTag<>("main-output"); + SystemReduceFn reduceFn = + SystemReduceFn.combining( + VarLongCoder.of(), + AppliedCombineFn.withInputCoder( + Sum.ofLongs(), + CoderRegistry.createDefault(), + KvCoder.of(VarLongCoder.of(), VarLongCoder.of()))); + Coder windowCoder = GlobalWindow.Coder.INSTANCE; + FullWindowedValueCoder> inputCoder = + WindowedValues.getFullCoder( + SingletonKeyedWorkItemCoder.of(VarLongCoder.of(), VarLongCoder.of(), windowCoder), + windowCoder); + FullWindowedValueCoder> outputCoder = + WindowedValues.getFullCoder(KvCoder.of(VarLongCoder.of(), VarLongCoder.of()), windowCoder); + FlinkPipelineOptions options = options(); + return new WindowDoFnOperator( + reduceFn, + STEP, + (Coder) inputCoder, + outputTag, + emptyList(), + new MultiOutputOutputManagerFactory<>( + outputTag, outputCoder, new SerializablePipelineOptions(options)), + windowingStrategy, + emptyMap(), + emptyList(), + options, + VarLongCoder.of(), + new WorkItemKeySelector(VarLongCoder.of())); + } + + private static SortedBatchCombineOperator sortedOperator( + SortedBatchCombineOperator.KeyCombiner combiner, + TimestampCombiner timestampCombiner, + FlinkPipelineOptions options) { + return new SortedBatchCombineOperator<>( + STEP, combiner, timestampCombiner, new SerializablePipelineOptions(options)); + } + + @Test + public void producesTheSameOutputAsWindowDoFnOperator() throws Exception { + for (TimestampCombiner timestampCombiner : TimestampCombiner.values()) { + KeyedOneInputStreamOperatorTestHarness< + FlinkKey, WindowedValue>, WindowedValue>> + expectedHarness = harness(windowDoFnOperator(timestampCombiner)); + expectedHarness.open(); + for (long[] e : sortedInput()) { + expectedHarness.processElement(element(e[0], e[1], e[2]), e[2]); + } + expectedHarness.processWatermark(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()); + List>> expected = new ArrayList<>(); + stripStreamRecordFromWindowedValue(expectedHarness.getOutput()) + .forEach(v -> expected.add((WindowedValue>) (WindowedValue) v)); + expectedHarness.close(); + + // The final step receives the accumulators of the partial combine. + KeyedOneInputStreamOperatorTestHarness< + FlinkKey, WindowedValue>, WindowedValue>> + harness = + harness( + sortedOperator( + new SortedBatchCombineOperator.MergingCombiner<>(Sum.ofLongs()), + timestampCombiner, + options())); + harness.open(); + for (long[] e : sortedInput()) { + harness.processElement(element(e[0], new long[] {e[1]}, e[2]), e[2]); + } + harness.endInput(); + List>> actual = new ArrayList<>(); + stripStreamRecordFromWindowedValue(harness.getOutput()) + .forEach(v -> actual.add((WindowedValue>) (WindowedValue) v)); + harness.close(); + + assertThat(timestampCombiner.toString(), expected.size(), is(3)); + assertThat(timestampCombiner.toString(), actual, containsInAnyOrder(expected.toArray())); + } + } + + @Test + public void bufferingCombinerKeepsAllAccumulatorsInOrder() throws Exception { + KeyedOneInputStreamOperatorTestHarness< + FlinkKey, + WindowedValue>>, + WindowedValue>>>> + harness = + harness( + sortedOperator( + new SortedBatchCombineOperator.BufferingCombiner>(), + TimestampCombiner.END_OF_WINDOW, + options())); + harness.open(); + harness.processElement(element(1L, Arrays.asList(1L, 2L), 0), 0); + harness.processElement(element(1L, Arrays.asList(3L), 0), 0); + harness.processElement(element(2L, Arrays.asList(4L), 0), 0); + harness.endInput(); + + List>>> actual = new ArrayList<>(); + for (Object o : stripStreamRecordFromWindowedValue(harness.getOutput())) { + KV>> kv = + ((WindowedValue>>>) o).getValue(); + List> lists = new ArrayList<>(); + kv.getValue().forEach(l -> lists.add(new ArrayList<>((List) l))); + actual.add(KV.of(kv.getKey(), lists)); + } + harness.close(); + + assertThat( + actual, + is( + Arrays.asList( + KV.of(1L, Arrays.asList(Arrays.asList(1L, 2L), Arrays.asList(3L))), + KV.of(2L, Arrays.asList(Arrays.asList(4L)))))); + } + + /** Sums accumulators and appends the job name, read from the combine context. */ + private static class JobNameSum + extends CombineWithContext.CombineFnWithContext, String> { + @Override + public List createAccumulator(CombineWithContext.Context c) { + return new ArrayList<>(); + } + + @Override + public List addInput(List accumulator, Long input, CombineWithContext.Context c) { + accumulator.add(input); + return accumulator; + } + + @Override + public List mergeAccumulators( + Iterable> accumulators, CombineWithContext.Context c) { + List merged = new ArrayList<>(); + accumulators.forEach(merged::addAll); + return merged; + } + + @Override + public String extractOutput(List accumulator, CombineWithContext.Context c) { + long sum = accumulator.stream().mapToLong(Long::longValue).sum(); + return c.getPipelineOptions().getJobName() + ":" + sum; + } + } + + @Test + public void providesPipelineOptionsToCombineFnsWithContext() throws Exception { + FlinkPipelineOptions options = options(); + options.setJobName("job"); + KeyedOneInputStreamOperatorTestHarness< + FlinkKey, WindowedValue>>, WindowedValue>> + harness = + harness( + sortedOperator( + new SortedBatchCombineOperator.MergingCombiner<>(new JobNameSum()), + TimestampCombiner.END_OF_WINDOW, + options)); + harness.open(); + harness.processElement(element(1L, new ArrayList<>(Arrays.asList(1L, 2L)), 0), 0); + harness.processElement(element(1L, new ArrayList<>(Arrays.asList(4L)), 0), 0); + harness.endInput(); + + List>> actual = new ArrayList<>(); + stripStreamRecordFromWindowedValue(harness.getOutput()) + .forEach(v -> actual.add((WindowedValue>) (WindowedValue) v)); + harness.close(); + + assertThat( + actual, + is( + Arrays.asList( + WindowedValues.of( + KV.of(1L, "job:7"), + GlobalWindow.INSTANCE.maxTimestamp(), + GlobalWindow.INSTANCE, + PaneInfo.ON_TIME_AND_ONLY_FIRING)))); + } + + /** A sum that counts how often accumulators are merged. */ + private static class CountingSum extends Combine.BinaryCombineLongFn { + @Override + public long apply(long left, long right) { + Metrics.counter("test", "merges").inc(); + return left + right; + } + + @Override + public long identity() { + return 0; + } + } + + @Test + public void reportsMetricsOfTheCombineFn() throws Exception { + SortedBatchCombineOperator operator = + sortedOperator( + new SortedBatchCombineOperator.MergingCombiner<>(new CountingSum()), + TimestampCombiner.END_OF_WINDOW, + options()); + KeyedOneInputStreamOperatorTestHarness< + FlinkKey, WindowedValue>, WindowedValue>> + harness = harness(operator); + harness.open(); + harness.processElement(element(1L, new long[] {1}, 0), 0); + harness.processElement(element(1L, new long[] {2}, 0), 0); + harness.endInput(); + + MetricsContainerImpl container = operator.metricContainer.getMetricsContainer(STEP); + assertThat( + container.getCounter(MetricName.named("test", "merges")).getCumulative() > 0, is(true)); + harness.close(); + } + + @Test + public void emitsTheLastKeyBeforeTheFinalWatermark() throws Exception { + KeyedOneInputStreamOperatorTestHarness< + FlinkKey, WindowedValue>, WindowedValue>> + harness = + harness( + sortedOperator( + new SortedBatchCombineOperator.MergingCombiner<>(Sum.ofLongs()), + TimestampCombiner.END_OF_WINDOW, + options())); + harness.open(); + harness.processElement(element(1L, new long[] {1}, 0), 0); + // Intermediate watermarks are held back, as the pending output may be behind them. + harness.processWatermark(10L); + harness.processElement(element(1L, new long[] {2}, 0), 0); + harness.processWatermark(Watermark.MAX_WATERMARK); + harness.endInput(); + + List output = new ArrayList<>(harness.getOutput()); + harness.close(); + + assertThat(output.size(), is(2)); + assertThat( + ((StreamRecord>>) output.get(0)).getValue().getValue(), + is(KV.of(1L, 3L))); + assertThat(output.get(1), is(Watermark.MAX_WATERMARK)); + } + + @Test + public void emitsNothingForEmptyInput() throws Exception { + KeyedOneInputStreamOperatorTestHarness< + FlinkKey, WindowedValue>, WindowedValue>> + harness = + harness( + sortedOperator( + new SortedBatchCombineOperator.MergingCombiner<>(Sum.ofLongs()), + TimestampCombiner.END_OF_WINDOW, + options())); + harness.open(); + harness.endInput(); + assertThat(stripStreamRecordFromWindowedValue(harness.getOutput()), emptyIterable()); + harness.close(); + } +} diff --git a/website/www/site/layouts/shortcodes/flink_java_pipeline_options.html b/website/www/site/layouts/shortcodes/flink_java_pipeline_options.html index 8e0fc007f13b..fad0dad62659 100644 --- a/website/www/site/layouts/shortcodes/flink_java_pipeline_options.html +++ b/website/www/site/layouts/shortcodes/flink_java_pipeline_options.html @@ -187,6 +187,11 @@ Shuts down sources which have been idle for the configured time of milliseconds. Once a source has been shut down, checkpointing is not possible anymore. Shutting down the sources eventually leads to pipeline shutdown (=Flink job finishes) once all input has been processed. Unless explicitly set, this will default to Long.MAX_VALUE when checkpointing is enabled and to 0 when checkpointing is disabled. See https://issues.apache.org/jira/browse/FLINK-2491 for progress on this issue. Default: -1 + + sortedBatchCombine + Batch only. Use a dedicated operator for the final step of GroupByKey and Combine in the global window with the default trigger. It relies on Flink sorting the input by key and does not use keyed state or timers. Has no effect in streaming mode. Enabled by default; set to false to disable. + Default: true + sourceStaticSplitThresholdMb Static split assignment threshold in MiB per source reader for bounded sources in Flink DataStream mode. The default of 0 always uses lazy assignment. A positive value selects static round-robin assignment for sources with a known, positive estimate below the threshold and lazy assignment otherwise. Any negative value always uses static assignment. diff --git a/website/www/site/layouts/shortcodes/flink_python_pipeline_options.html b/website/www/site/layouts/shortcodes/flink_python_pipeline_options.html index d2ab687c42be..6bacbc826136 100644 --- a/website/www/site/layouts/shortcodes/flink_python_pipeline_options.html +++ b/website/www/site/layouts/shortcodes/flink_python_pipeline_options.html @@ -187,6 +187,11 @@ Shuts down sources which have been idle for the configured time of milliseconds. Once a source has been shut down, checkpointing is not possible anymore. Shutting down the sources eventually leads to pipeline shutdown (=Flink job finishes) once all input has been processed. Unless explicitly set, this will default to Long.MAX_VALUE when checkpointing is enabled and to 0 when checkpointing is disabled. See https://issues.apache.org/jira/browse/FLINK-2491 for progress on this issue. Default: -1 + + sorted_batch_combine + Batch only. Use a dedicated operator for the final step of GroupByKey and Combine in the global window with the default trigger. It relies on Flink sorting the input by key and does not use keyed state or timers. Has no effect in streaming mode. Enabled by default; set to false to disable. + Default: true + source_static_split_threshold_mb Static split assignment threshold in MiB per source reader for bounded sources in Flink DataStream mode. The default of 0 always uses lazy assignment. A positive value selects static round-robin assignment for sources with a known, positive estimate below the threshold and lazy assignment otherwise. Any negative value always uses static assignment.