You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ec...@apache.org on 2019/07/09 13:18:17 UTC
[beam] branch spark-runner_structured-streaming updated (c6d2c52 ->
a37817d)
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a change to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git.
discard c6d2c52 Add a test to check that CombineGlobally preserves windowing
discard f125725 Fix accumulators initialization in Combine that prevented CombineGlobally to work.
discard 44e851d Fix javadoc
discard d14a29a Add setEnableSparkMetricSinks() method
discard 7356ec2 Add missing dependencies to run Spark Structured Streaming Runner on Nexmark
discard d5f322b Add metrics support in DoFn
discard 7803025 Ignore for now not working test testCombineGlobally
discard 996a296 Add a test that combine per key preserves windowing
discard 418df85 Clean groupByKeyTest
discard 41a2027 add comment in combine globally test
discard 19779af Fixed immutable list bug
discard 4110d8f Fix javadoc of AggregatorCombiner
discard 319f4c9 Clean not more needed WindowingHelpers
discard a0cd463 Clean not more needed RowHelpers
discard 930974f Clean no more needed KVHelpers
discard 1c28c63 Now that there is only Combine.PerKey translation, make only one Aggregator
discard 8569c9d Remove CombineGlobally translation because it is less performant than the beam sdk one (key + combinePerKey.withHotkeyFanout)
discard d869c50 Remove the mapPartition that adds a key per partition because otherwise spark will reduce values per key instead of globally
discard aa5543f Fix bug in the window merging logic
discard 5a5ca48 Fix wrong encoder in combineGlobally GBK
discard ac06721 Fix case when a window does not merge into any other window
discard a210466 Apply a groupByKey avoids for some reason that the spark structured streaming fmwk casts data to Row which makes it impossible to deserialize without the coder shipped into the data. For performance reasons (avoid memory consumption and having to deserialize), we do not ship coder + data. Also add a mapparitions before GBK to avoid shuffling
discard f4f186b Revert extractKey while combinePerKey is not done (so that it compiles)
discard c4e3f44 Fix encoder in combine call
discard a269fa26 Implement merge accumulators part of CombineGlobally translation with windowing
discard 49293d7 Output data after combine
discard dea9155 Implement reduce part of CombineGlobally translation with windowing
discard 50dac73 Fix comment about schemas
discard 6a6b959 Update KVHelpers.extractKey() to deal with WindowedValue and update GBK and CPK
discard 2c16deb Add TODO in Combine translations
discard 0c30d20 Add a test that GBK preserves windowing
discard 8dad2a7 fixup Enable UsesBoundedSplittableParDo category of tests Since the default overwrite is already in place and it is breaking only in one subcase
discard 4d75434 Improve visibility of debug messages
discard 32756ef re-enable reduceFnRunner timers for output
discard d2cebcc Re-code GroupByKeyTranslatorBatch to conserve windowing instead of unwindowing/windowing(GlobalWindow): simplify code, use ReduceFnRunner to merge the windows
discard ff937d9 Add comment about checkpoint mark
discard f0c415c Update windowAssignTest
discard 8ee2e60 Put back batch/simpleSourceTest.testBoundedSource
discard a5ca4df Consider null object case on RowHelpers, fixes empty side inputs tests.
discard b0f9afd fixup hadoop-format is not mandataory to run ValidatesRunner tests
discard e2eb99a Pass transform based doFnSchemaInformation in ParDo translation
discard 785c0d9 fixup Enable UsesFailureMessage and UsesSchema categories on validatesRunner tests
discard d551903 Fixes ParDo not calling setup and not tearing down if exception on startBundle
discard 20d5a68 Limit the number of partitions to make tests go 300% faster
discard a9bdab6 Add Batch Validates Runner tests for Structured Streaming Runner
new 82e28d1 Build fixup: enable batch Validates Runner tests for Structured Streaming Runner
new 1d52f35 Build fixup: Limit the number of partitions to make tests go 300% faster
new 1f4f0bc Fixes ParDo not calling setup and not tearing down if exception on startBundle
new ff50759 Build fixup: enable UsesFailureMessage and UsesSchema categories on validatesRunner tests
new cb1158b Pass transform based doFnSchemaInformation in ParDo translation
new ebc6ca9 Build fixup: hadoop-format is not mandataory to run ValidatesRunner tests
new 9c3b009 Consider null object case on RowHelpers, fixes empty side inputs tests.
new 9d29d76 Put back batch/simpleSourceTest.testBoundedSource
new ab26efc Update windowAssignTest
new ede9118 Add comment about checkpoint mark
new 6bc9b17 Re-code GroupByKeyTranslatorBatch to conserve windowing instead of unwindowing/windowing(GlobalWindow): simplify code, use ReduceFnRunner to merge the windows
new 6c9a3a8 re-enable reduceFnRunner timers for output
new ee91d24 Improve visibility of debug messages
new 41de6da Build fixup: enable UsesBoundedSplittableParDo category of tests Since the default overwrite is already in place and it is breaking only in one subcase
new 92bf293 Add a test that GBK preserves windowing
new 1a85496 Add TODO in Combine translations
new 05cb7b8 Update KVHelpers.extractKey() to deal with WindowedValue and update GBK and CPK
new a9e7bf4 Fix comment about schemas
new 233e532 Implement reduce part of CombineGlobally translation with windowing
new 3e49bf0 Output data after combine
new a4af031 Implement merge accumulators part of CombineGlobally translation with windowing
new 28067b2 Fix encoder in combine call
new b1cf536 Revert extractKey while combinePerKey is not done (so that it compiles)
new e77e951 Apply a groupByKey avoids for some reason that the spark structured streaming fmwk casts data to Row which makes it impossible to deserialize without the coder shipped into the data. For performance reasons (avoid memory consumption and having to deserialize), we do not ship coder + data. Also add a mapparitions before GBK to avoid shuffling
new 3ae7e09 Fix case when a window does not merge into any other window
new ddce346 Fix wrong encoder in combineGlobally GBK
new 11981e5 Fix bug in the window merging logic
new 6450bda Remove the mapPartition that adds a key per partition because otherwise spark will reduce values per key instead of globally
new 18a2e2a Remove CombineGlobally translation because it is less performant than the beam sdk one (key + combinePerKey.withHotkeyFanout)
new f05e17c Now that there is only Combine.PerKey translation, make only one Aggregator
new 2851c72 Clean no more needed KVHelpers
new e84c0bb Clean not more needed RowHelpers
new ea959c6 Clean not more needed WindowingHelpers
new 721b1d0 Fix javadoc of AggregatorCombiner
new 78270bf Fixed immutable list bug
new 777be24 add comment in combine globally test
new bea2c60 Clean groupByKeyTest
new 5bb44d6 Add a test that combine per key preserves windowing
new 703e773 Ignore for now not working test testCombineGlobally
new 12eb126 Add metrics support in DoFn
new f88db1b Add missing dependencies to run Spark Structured Streaming Runner on Nexmark
new a3d1061 Add setEnableSparkMetricSinks() method
new 0fc97ee Fix javadoc
new 170568b Fix accumulators initialization in Combine that prevented CombineGlobally to work.
new a37817d Add a test to check that CombineGlobally preserves windowing
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (c6d2c52)
\
N -- N -- N refs/heads/spark-runner_structured-streaming (a37817d)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 45 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
[beam] 20/45: Output data after combine
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 3e49bf0bbc57e9e41155271b4082d8a887673232
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Fri Jun 14 14:50:45 2019 +0200
Output data after combine
---
.../translation/batch/AggregatorCombinerGlobally.java | 14 ++++++++------
.../translation/batch/CombineGloballyTranslatorBatch.java | 7 +++++--
2 files changed, 13 insertions(+), 8 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
index 0d13218..6996165 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
@@ -44,7 +44,7 @@ import scala.Tuple2;
* */
class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindow>
- extends Aggregator<WindowedValue<InputT>, Iterable<WindowedValue<AccumT>>, WindowedValue<OutputT>> {
+ extends Aggregator<WindowedValue<InputT>, Iterable<WindowedValue<AccumT>>, Iterable<WindowedValue<OutputT>>> {
private final Combine.CombineFn<InputT, AccumT, OutputT> combineFn;
private WindowingStrategy<InputT, W> windowingStrategy;
@@ -123,10 +123,12 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
return null;
}
- @Override public WindowedValue<OutputT> finish(Iterable<WindowedValue<AccumT>> reduction) {
- // TODO
- // return combineFn.extractOutput(reduction);
- return null;
+ @Override public Iterable<WindowedValue<OutputT>> finish(Iterable<WindowedValue<AccumT>> reduction) {
+ List<WindowedValue<OutputT>> result = new ArrayList<>();
+ for (WindowedValue<AccumT> windowedValue: reduction) {
+ result.add(windowedValue.withValue(combineFn.extractOutput(windowedValue.getValue())));
+ }
+ return result;
}
@Override public Encoder<Iterable<WindowedValue<AccumT>>> bufferEncoder() {
@@ -134,7 +136,7 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
return EncoderHelpers.genericEncoder();
}
- @Override public Encoder<WindowedValue<OutputT>> outputEncoder() {
+ @Override public Encoder<Iterable<WindowedValue<OutputT>>> outputEncoder() {
// TODO replace with outputCoder if possible
return EncoderHelpers.genericEncoder();
}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
index f18572b..fb9e1dd 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
@@ -21,12 +21,12 @@ import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTr
import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.RowHelpers;
-import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.WindowingHelpers;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.WindowingStrategy;
+import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
@@ -57,9 +57,12 @@ class CombineGloballyTranslatorBatch<InputT, AccumT, OutputT>
Dataset<Row> combinedRowDataset =
inputDataset.agg(new AggregatorCombinerGlobally<>(combineFn, windowingStrategy).toColumn());
- Dataset<WindowedValue<OutputT>> outputDataset =
+ Dataset<Iterable<WindowedValue<OutputT>>> accumulatedDataset =
combinedRowDataset.map(
RowHelpers.extractObjectFromRowMapFunction(), EncoderHelpers.windowedValueEncoder());
+ Dataset<WindowedValue<OutputT>> outputDataset = accumulatedDataset.flatMap(
+ (FlatMapFunction<Iterable<WindowedValue<OutputT>>, WindowedValue<OutputT>>)
+ windowedValues -> windowedValues.iterator(), EncoderHelpers.windowedValueEncoder());
context.putDataset(output, outputDataset);
}
}
[beam] 30/45: Now that there is only Combine.PerKey translation,
make only one Aggregator
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit f05e17c49ad4b3c004328ea18505ed7848ffa318
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 11:20:12 2019 +0200
Now that there is only Combine.PerKey translation, make only one Aggregator
---
...mbinerGlobally.java => AggregatorCombiner.java} | 20 ++++++-----
.../batch/CombinePerKeyTranslatorBatch.java | 41 +++++++++++++---------
2 files changed, 35 insertions(+), 26 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
similarity index 92%
rename from runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
rename to runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
index d4c35d8..c61d937 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
@@ -33,6 +33,7 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
import org.apache.spark.sql.Encoder;
@@ -45,14 +46,14 @@ import scala.Tuple2;
* The accumulator is a {@code Iterable<WindowedValue<AccumT>> because an {@code InputT} can be in multiple windows. So, when accumulating {@code InputT} values, we create one accumulator per input window.
* */
-class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindow>
- extends Aggregator<WindowedValue<InputT>, Iterable<WindowedValue<AccumT>>, Iterable<WindowedValue<OutputT>>> {
+class AggregatorCombiner<K, InputT, AccumT, OutputT, W extends BoundedWindow>
+ extends Aggregator<WindowedValue<KV<K, InputT>>, Iterable<WindowedValue<AccumT>>, Iterable<WindowedValue<OutputT>>> {
private final Combine.CombineFn<InputT, AccumT, OutputT> combineFn;
private WindowingStrategy<InputT, W> windowingStrategy;
private TimestampCombiner timestampCombiner;
- public AggregatorCombinerGlobally(Combine.CombineFn<InputT, AccumT, OutputT> combineFn, WindowingStrategy<?, ?> windowingStrategy) {
+ public AggregatorCombiner(Combine.CombineFn<InputT, AccumT, OutputT> combineFn, WindowingStrategy<?, ?> windowingStrategy) {
this.combineFn = combineFn;
this.windowingStrategy = (WindowingStrategy<InputT, W>) windowingStrategy;
this.timestampCombiner = windowingStrategy.getTimestampCombiner();
@@ -63,10 +64,11 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
}
@Override public Iterable<WindowedValue<AccumT>> reduce(Iterable<WindowedValue<AccumT>> accumulators,
- WindowedValue<InputT> input) {
+ WindowedValue<KV<K, InputT>> inputWv) {
+ KV<K, InputT> inputKv = inputWv.getValue();
//concatenate accumulators windows and input windows and merge the windows
- Collection<W> inputWindows = (Collection<W>)input.getWindows();
+ Collection<W> inputWindows = (Collection<W>)inputWv.getWindows();
Set<W> windows = collectAccumulatorsWindows(accumulators);
windows.addAll(inputWindows);
Map<W, W> windowToMergeResult;
@@ -86,17 +88,17 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
Tuple2<AccumT, Instant> accumAndInstant = windowToAccumAndInstant.get(mergedWindow);
// if there is no accumulator associated with this window yet, create one
if (accumAndInstant == null) {
- AccumT accum = combineFn.addInput(combineFn.createAccumulator(), input.getValue());
+ AccumT accum = combineFn.addInput(combineFn.createAccumulator(), inputKv.getValue());
Instant windowTimestamp =
timestampCombiner.assign(
- mergedWindow, windowingStrategy.getWindowFn().getOutputTime(input.getTimestamp(), mergedWindow));
+ mergedWindow, windowingStrategy.getWindowFn().getOutputTime(inputWv.getTimestamp(), mergedWindow));
accumAndInstant = new Tuple2<>(accum, windowTimestamp);
} else {
AccumT updatedAccum =
- combineFn.addInput(accumAndInstant._1, input.getValue());
+ combineFn.addInput(accumAndInstant._1, inputKv.getValue());
Instant updatedTimestamp = timestampCombiner.combine(accumAndInstant._2, timestampCombiner
.assign(mergedWindow,
- windowingStrategy.getWindowFn().getOutputTime(input.getTimestamp(), mergedWindow)));
+ windowingStrategy.getWindowFn().getOutputTime(inputWv.getTimestamp(), mergedWindow)));
accumAndInstant = new Tuple2<>(updatedAccum, updatedTimestamp);
}
windowToAccumAndInstant.put(mergedWindow, accumAndInstant);
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
index 1c35301..12e5944 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
@@ -17,18 +17,20 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.AggregatorCombinerPerKey;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.KVHelpers;
-import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.WindowingHelpers;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.spark.api.java.function.MapFunction;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.KeyValueGroupedDataset;
import scala.Tuple2;
@@ -50,25 +52,30 @@ class CombinePerKeyTranslatorBatch<K, InputT, AccumT, OutputT>
@SuppressWarnings("unchecked")
final Combine.CombineFn<InputT, AccumT, OutputT> combineFn =
(Combine.CombineFn<InputT, AccumT, OutputT>) combineTransform.getFn();
+ WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
Dataset<WindowedValue<KV<K, InputT>>> inputDataset = context.getDataset(input);
- Dataset<KV<K, InputT>> unwindowedDataset = inputDataset
- .map(WindowingHelpers.unwindowMapFunction(), EncoderHelpers.kvEncoder());
- KeyValueGroupedDataset<K, KV<K, InputT>> groupedDataset =
- unwindowedDataset.groupByKey((MapFunction<KV<K, InputT>, K>) kv -> kv.getKey(), EncoderHelpers.genericEncoder());
+ KeyValueGroupedDataset<K, WindowedValue<KV<K, InputT>>> groupedDataset = inputDataset
+ .groupByKey(KVHelpers.extractKey(), EncoderHelpers.genericEncoder());
- Dataset<Tuple2<K, OutputT>> combinedDataset =
- groupedDataset.agg(
- new AggregatorCombinerPerKey<K, InputT, AccumT, OutputT>(combineFn).toColumn());
+ Dataset<Tuple2<K, Iterable<WindowedValue<OutputT>>>> combinedDataset = groupedDataset.agg(
+ new AggregatorCombiner<K, InputT, AccumT, OutputT, BoundedWindow>(combineFn,
+ windowingStrategy).toColumn());
- Dataset<KV<K, OutputT>> kvOutputDataset =
- combinedDataset.map(KVHelpers.tuple2ToKV(), EncoderHelpers.kvEncoder());
-
- // Window the result into global window.
- Dataset<WindowedValue<KV<K, OutputT>>> outputDataset =
- kvOutputDataset.map(
- WindowingHelpers.windowMapFunction(), EncoderHelpers.windowedValueEncoder());
+ Dataset<WindowedValue<KV<K, OutputT>>> outputDataset = combinedDataset.flatMap(
+ (FlatMapFunction<Tuple2<K, Iterable<WindowedValue<OutputT>>>, WindowedValue<KV<K, OutputT>>>) tuple2 -> {
+ K key = tuple2._1;
+ Iterable<WindowedValue<OutputT>> windowedValues = tuple2._2;
+ List<WindowedValue<KV<K, OutputT>>> result = new ArrayList<>();
+ for (WindowedValue<OutputT> windowedValue : windowedValues) {
+ KV<K, OutputT> kv = KV.of(key, windowedValue.getValue());
+ result.add(WindowedValue
+ .of(kv, windowedValue.getTimestamp(), windowedValue.getWindows(),
+ windowedValue.getPane()));
+ }
+ return result.iterator();
+ }, EncoderHelpers.windowedValueEncoder());
context.putDataset(output, outputDataset);
}
}
[beam] 18/45: Fix comment about schemas
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a9e7bf451cead9b76f35811fb76362ed3cddda7c
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue Jun 11 14:40:54 2019 +0200
Fix comment about schemas
---
.../batch/AggregatorCombinerGlobally.java | 2 +-
.../translation/helpers/ReduceFnRunnerHelpers.java | 77 ++++++++++++++++++++++
2 files changed, 78 insertions(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
index a03c17e..2f8293b 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
@@ -43,7 +43,7 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT>
@Override
public AccumT reduce(AccumT accumulator, InputT input) {
// because of generic type InputT, spark cannot infer an input type.
- // it would pass Integer as input if we had a Aggregator<Input, ..., ...>
+ // it would pass Integer as input if we had a Aggregator<Integer, ..., ...>
// without the type inference it stores input in a GenericRowWithSchema
Row row = (Row) input;
InputT t = RowHelpers.extractObjectFromRow(row);
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/ReduceFnRunnerHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/ReduceFnRunnerHelpers.java
new file mode 100644
index 0000000..97a225e
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/ReduceFnRunnerHelpers.java
@@ -0,0 +1,77 @@
+package org.apache.beam.runners.spark.structuredstreaming.translation.helpers;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import org.apache.beam.runners.core.InMemoryTimerInternals;
+import org.apache.beam.runners.core.OutputWindowedValue;
+import org.apache.beam.runners.core.ReduceFnRunner;
+import org.apache.beam.runners.core.TimerInternals;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.TupleTag;
+import org.joda.time.Instant;
+
+/**
+ * Helpers to use {@link ReduceFnRunner}.
+ */
+public class ReduceFnRunnerHelpers<K, InputT, W extends BoundedWindow> {
+ public static <K, InputT, W extends BoundedWindow> void fireEligibleTimers(
+ InMemoryTimerInternals timerInternals,
+ ReduceFnRunner<K, InputT, Iterable<InputT>, W> reduceFnRunner)
+ throws Exception {
+ List<TimerInternals.TimerData> timers = new ArrayList<>();
+ while (true) {
+ TimerInternals.TimerData timer;
+ while ((timer = timerInternals.removeNextEventTimer()) != null) {
+ timers.add(timer);
+ }
+ while ((timer = timerInternals.removeNextProcessingTimer()) != null) {
+ timers.add(timer);
+ }
+ while ((timer = timerInternals.removeNextSynchronizedProcessingTimer()) != null) {
+ timers.add(timer);
+ }
+ if (timers.isEmpty()) {
+ break;
+ }
+ reduceFnRunner.onTimers(timers);
+ timers.clear();
+ }
+ }
+
+ /**
+ * {@link OutputWindowedValue} for ReduceFnRunner.
+ *
+ */
+ public static class GABWOutputWindowedValue<K, V>
+ implements OutputWindowedValue<KV<K, Iterable<V>>> {
+ private final List<WindowedValue<KV<K, Iterable<V>>>> outputs = new ArrayList<>();
+
+ @Override
+ public void outputWindowedValue(
+ KV<K, Iterable<V>> output,
+ Instant timestamp,
+ Collection<? extends BoundedWindow> windows,
+ PaneInfo pane) {
+ outputs.add(WindowedValue.of(output, timestamp, windows, pane));
+ }
+
+ @Override
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
+ Instant timestamp,
+ Collection<? extends BoundedWindow> windows,
+ PaneInfo pane) {
+ throw new UnsupportedOperationException("GroupAlsoByWindow should not use tagged outputs.");
+ }
+
+ public Iterable<WindowedValue<KV<K, Iterable<V>>>> getOutputs() {
+ return outputs;
+ }
+ }
+
+}
[beam] 44/45: Fix accumulators initialization in Combine that
prevented CombineGlobally to work.
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 170568b4f45bb41ada82299a154e032bc3c77cff
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue Jul 9 14:44:54 2019 +0200
Fix accumulators initialization in Combine that prevented CombineGlobally to work.
---
.../translation/batch/AggregatorCombiner.java | 61 ++++------------------
.../batch/CombinePerKeyTranslatorBatch.java | 1 +
.../translation/batch/CombineTest.java | 1 -
3 files changed, 11 insertions(+), 52 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
index e005c6f..001fe43 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
@@ -39,8 +39,6 @@ import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Lists;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.expressions.Aggregator;
-import org.joda.time.Instant;
-import scala.Tuple2;
/** An {@link Aggregator} for the Spark Batch Runner.
* The accumulator is a {@code Iterable<WindowedValue<AccumT>> because an {@code InputT} can be in multiple windows. So, when accumulating {@code InputT} values, we create one accumulator per input window.
@@ -63,56 +61,16 @@ class AggregatorCombiner<K, InputT, AccumT, OutputT, W extends BoundedWindow>
return new ArrayList<>();
}
+ private Iterable<WindowedValue<AccumT>> createAccumulator(WindowedValue<KV<K, InputT>> inputWv) {
+ AccumT accumulator = combineFn.createAccumulator();
+ combineFn.addInput(accumulator, inputWv.getValue().getValue());
+ return Lists.newArrayList(WindowedValue
+ .of(accumulator, inputWv.getTimestamp(), inputWv.getWindows(),
+ inputWv.getPane()));
+ }
@Override public Iterable<WindowedValue<AccumT>> reduce(Iterable<WindowedValue<AccumT>> accumulators,
WindowedValue<KV<K, InputT>> inputWv) {
-
- KV<K, InputT> inputKv = inputWv.getValue();
- //concatenate accumulators windows and input windows and merge the windows
- Collection<W> inputWindows = (Collection<W>)inputWv.getWindows();
- Set<W> windows = collectAccumulatorsWindows(accumulators);
- windows.addAll(inputWindows);
- Map<W, W> windowToMergeResult;
- try {
- windowToMergeResult = mergeWindows(windowingStrategy, windows);
- } catch (Exception e) {
- throw new RuntimeException("Unable to merge accumulators windows and input windows", e);
- }
-
- // iterate through the input windows and for each, create an accumulator with the merged window
- // associated to it and call addInput with the accumulator.
- // Maintain a map of the accumulators for use as output
- Map<W, Tuple2<AccumT, Instant>> windowToAccumAndInstant = new HashMap<>();
- for (W inputWindow:inputWindows) {
- W mergedWindow = windowToMergeResult.get(inputWindow);
- mergedWindow = mergedWindow == null ? inputWindow : mergedWindow;
- Tuple2<AccumT, Instant> accumAndInstant = windowToAccumAndInstant.get(mergedWindow);
- // if there is no accumulator associated with this window yet, create one
- if (accumAndInstant == null) {
- AccumT accum = combineFn.addInput(combineFn.createAccumulator(), inputKv.getValue());
- Instant windowTimestamp =
- timestampCombiner.assign(
- mergedWindow, windowingStrategy.getWindowFn().getOutputTime(inputWv.getTimestamp(), mergedWindow));
- accumAndInstant = new Tuple2<>(accum, windowTimestamp);
- } else {
- AccumT updatedAccum =
- combineFn.addInput(accumAndInstant._1, inputKv.getValue());
- Instant updatedTimestamp = timestampCombiner.combine(accumAndInstant._2, timestampCombiner
- .assign(mergedWindow,
- windowingStrategy.getWindowFn().getOutputTime(inputWv.getTimestamp(), mergedWindow)));
- accumAndInstant = new Tuple2<>(updatedAccum, updatedTimestamp);
- }
- windowToAccumAndInstant.put(mergedWindow, accumAndInstant);
- }
- // output the accumulators map
- //TODO perf: to avoid memory allocation it is authorized to modify accumulators argument
- List<WindowedValue<AccumT>> result = new ArrayList<>();
- for (Map.Entry<W, Tuple2<AccumT, Instant>> entry : windowToAccumAndInstant.entrySet()) {
- AccumT accumulator = entry.getValue()._1;
- Instant windowTimestamp = entry.getValue()._2;
- W window = entry.getKey();
- result.add(WindowedValue.of(accumulator, windowTimestamp, window, PaneInfo.NO_FIRING));
- }
- return result;
+ return merge(accumulators, createAccumulator(inputWv));
}
@Override public Iterable<WindowedValue<AccumT>> merge(
@@ -182,7 +140,8 @@ class AggregatorCombiner<K, InputT, AccumT, OutputT, W extends BoundedWindow>
// an accumulator has only one window associated to it.
W accumulatorWindow = (W) accumulator.getWindows().iterator().next();
windows.add(accumulatorWindow);
- } return windows;
+ }
+ return windows;
}
private Map<W, W> mergeWindows(WindowingStrategy<InputT, W> windowingStrategy, Set<W> windows)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
index 12e5944..6ba28ea 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
@@ -63,6 +63,7 @@ class CombinePerKeyTranslatorBatch<K, InputT, AccumT, OutputT>
new AggregatorCombiner<K, InputT, AccumT, OutputT, BoundedWindow>(combineFn,
windowingStrategy).toColumn());
+ //expand the list into separate elements and put the key back into the elements
Dataset<WindowedValue<KV<K, OutputT>>> outputDataset = combinedDataset.flatMap(
(FlatMapFunction<Tuple2<K, Iterable<WindowedValue<OutputT>>>, WindowedValue<KV<K, OutputT>>>) tuple2 -> {
K key = tuple2._1;
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
index ef4468b..3505da5 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
@@ -52,7 +52,6 @@ public class CombineTest implements Serializable {
p = Pipeline.create(options);
}
- @Ignore
@Test
public void testCombineGlobally() {
PCollection<Integer> input =
[beam] 03/45: Fixes ParDo not calling setup and not tearing down if
exception on startBundle
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 1f4f0bcbf40a8443e4c67f016b7ec4891777cd95
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Thu May 9 16:20:08 2019 +0200
Fixes ParDo not calling setup and not tearing down if exception on startBundle
---
runners/spark/build.gradle | 1 -
.../translation/batch/DoFnFunction.java | 6 ++++++
.../translation/batch/ProcessContext.java | 22 +++++++++++-----------
3 files changed, 17 insertions(+), 12 deletions(-)
diff --git a/runners/spark/build.gradle b/runners/spark/build.gradle
index 72e6013..ad43212 100644
--- a/runners/spark/build.gradle
+++ b/runners/spark/build.gradle
@@ -186,7 +186,6 @@ task validatesStructuredStreamingRunnerBatch(type: Test) {
includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner'
excludeCategories 'org.apache.beam.sdk.testing.UsesCustomWindowMerging'
excludeCategories 'org.apache.beam.sdk.testing.UsesFailureMessage'
- excludeCategories 'org.apache.beam.sdk.testing.UsesParDoLifecycle'
// Unbounded
excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedPCollections'
excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream'
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnFunction.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnFunction.java
index 4f158bb..daa802d 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnFunction.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnFunction.java
@@ -31,6 +31,7 @@ import org.apache.beam.runners.spark.structuredstreaming.translation.utils.Cache
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
@@ -54,6 +55,7 @@ public class DoFnFunction<InputT, OutputT>
implements MapPartitionsFunction<WindowedValue<InputT>, Tuple2<TupleTag<?>, WindowedValue<?>>> {
private final DoFn<InputT, OutputT> doFn;
+ private transient boolean wasSetupCalled;
private final WindowingStrategy<?, ?> windowingStrategy;
private final Map<PCollectionView<?>, WindowingStrategy<?, ?>> sideInputs;
private final SerializablePipelineOptions serializableOptions;
@@ -90,6 +92,10 @@ public class DoFnFunction<InputT, OutputT>
@Override
public Iterator<Tuple2<TupleTag<?>, WindowedValue<?>>> call(Iterator<WindowedValue<InputT>> iter)
throws Exception {
+ if (!wasSetupCalled && iter.hasNext()) {
+ DoFnInvokers.tryInvokeSetupFor(doFn);
+ wasSetupCalled = true;
+ }
DoFnOutputManager outputManager = new DoFnOutputManager();
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ProcessContext.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ProcessContext.java
index 33ca3d0..d3453a7 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ProcessContext.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ProcessContext.java
@@ -93,18 +93,18 @@ class ProcessContext<FnInputT, FnOutputT, OutputT> {
@Override
protected OutputT computeNext() {
- // Process each element from the (input) iterator, which produces, zero, one or more
- // output elements (of type V) in the output iterator. Note that the output
- // collection (and iterator) is reset between each call to processElement, so the
- // collection only holds the output values for each call to processElement, rather
- // than for the whole partition (which would use too much memory).
- if (!isBundleStarted) {
- isBundleStarted = true;
- // call startBundle() before beginning to process the partition.
- doFnRunner.startBundle();
- }
-
try {
+ // Process each element from the (input) iterator, which produces, zero, one or more
+ // output elements (of type V) in the output iterator. Note that the output
+ // collection (and iterator) is reset between each call to processElement, so the
+ // collection only holds the output values for each call to processElement, rather
+ // than for the whole partition (which would use too much memory).
+ if (!isBundleStarted) {
+ isBundleStarted = true;
+ // call startBundle() before beginning to process the partition.
+ doFnRunner.startBundle();
+ }
+
while (true) {
if (outputIterator.hasNext()) {
return outputIterator.next();
[beam] 06/45: Build fixup: hadoop-format is not mandataory to run
ValidatesRunner tests
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit ebc6ca935c8b1b7b5c64a2e64378f5ffdc86885a
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Fri May 10 11:36:23 2019 +0200
Build fixup: hadoop-format is not mandataory to run ValidatesRunner tests
---
runners/spark/build.gradle | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/runners/spark/build.gradle b/runners/spark/build.gradle
index 46e5c25..9619c5e 100644
--- a/runners/spark/build.gradle
+++ b/runners/spark/build.gradle
@@ -178,7 +178,9 @@ task validatesStructuredStreamingRunnerBatch(type: Test) {
systemProperty "spark.ui.showConsoleProgress", "false"
classpath = configurations.validatesRunner
- testClassesDirs = files(project(":beam-sdks-java-core").sourceSets.test.output.classesDirs) + files(project(":beam-sdks-java-io-hadoop-format").sourceSets.test.output.classesDirs) + files(project.sourceSets.test.output.classesDirs)
+ testClassesDirs += files(project(":beam-sdks-java-core").sourceSets.test.output.classesDirs)
+ testClassesDirs += files(project.sourceSets.test.output.classesDirs)
+
// Only one SparkContext may be running in a JVM (SPARK-2243)
forkEvery 1
maxParallelForks 4
[beam] 13/45: Improve visibility of debug messages
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit ee91d24ec5f99929503bc7ba250133075f489e67
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue May 28 09:31:59 2019 +0200
Improve visibility of debug messages
---
.../spark/structuredstreaming/translation/TranslationContext.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
index 4d17120..411aec3 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
@@ -201,7 +201,7 @@ public class TranslationContext {
// code.
List<WindowedValue> windowedValues = ((Dataset<WindowedValue>) dataset).collectAsList();
for (WindowedValue windowedValue : windowedValues) {
- LOG.debug(windowedValue.toString());
+ LOG.debug("**** dataset content {} ****", windowedValue.toString());
}
} else {
// apply a dummy fn just to apply for each action that will trigger the pipeline run in
[beam] 28/45: Remove the mapPartition that adds a key per partition
because otherwise spark will reduce values per key instead of globally
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 6450bda372a61b9994444ac9f1ce1fdbbc952def
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed Jul 3 16:52:57 2019 +0200
Remove the mapPartition that adds a key per partition because otherwise spark will reduce values per key instead of globally
---
.../translation/batch/AggregatorCombinerGlobally.java | 5 ++---
.../batch/CombineGloballyTranslatorBatch.java | 19 ++-----------------
2 files changed, 4 insertions(+), 20 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
index 771e42c..d4c35d8 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
@@ -46,7 +46,7 @@ import scala.Tuple2;
* */
class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindow>
- extends Aggregator<Tuple2<Integer, WindowedValue<InputT>>, Iterable<WindowedValue<AccumT>>, Iterable<WindowedValue<OutputT>>> {
+ extends Aggregator<WindowedValue<InputT>, Iterable<WindowedValue<AccumT>>, Iterable<WindowedValue<OutputT>>> {
private final Combine.CombineFn<InputT, AccumT, OutputT> combineFn;
private WindowingStrategy<InputT, W> windowingStrategy;
@@ -63,9 +63,8 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
}
@Override public Iterable<WindowedValue<AccumT>> reduce(Iterable<WindowedValue<AccumT>> accumulators,
- Tuple2<Integer, WindowedValue<InputT>> inputTuple) {
+ WindowedValue<InputT> input) {
- WindowedValue<InputT> input = inputTuple._2;
//concatenate accumulators windows and input windows and merge the windows
Collection<W> inputWindows = (Collection<W>)input.getWindows();
Set<W> windows = collectAccumulatorsWindows(accumulators);
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
index 1a77fc3..30101dc 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
@@ -64,23 +64,8 @@ class CombineGloballyTranslatorBatch<InputT, AccumT, OutputT>
// the coder shipped into the data. For performance reasons
// (avoid memory consumption and having to deserialize), we do not ship coder + data.
- // We do not want to shuffle data during groupByKey, we cannot get the number of partitions for
- // the input dataset without triggering a costly operation (conversion to rdd) so we cannot use spark Hashpartitioner
- // so we apply a key to each input dataset partition and then trigger a GBK that should not shuffle data.
-
- Dataset<Tuple2<Integer, WindowedValue<InputT>>> keyedDataset = inputDataset
- .mapPartitions((MapPartitionsFunction<WindowedValue<InputT>, Tuple2<Integer, WindowedValue<InputT>>>) inputTIterator -> {
- List<Tuple2<Integer, WindowedValue<InputT>>> result = new ArrayList<>();
- Random random = new Random();
- while (inputTIterator.hasNext()) {
- result.add(Tuple2.apply(random.nextInt(), inputTIterator.next()));
- }
- return result.iterator();
- }, EncoderHelpers.tuple2Encoder());
-
- KeyValueGroupedDataset<Integer, Tuple2<Integer, WindowedValue<InputT>>> groupedDataset = keyedDataset
- .groupByKey(
- (MapFunction<Tuple2<Integer, WindowedValue<InputT>>, Integer>) value -> value._1(),
+ KeyValueGroupedDataset<Integer, WindowedValue<InputT>> groupedDataset = inputDataset
+ .groupByKey((MapFunction<WindowedValue<InputT>, Integer>) value -> 1,
EncoderHelpers.genericEncoder());
Dataset<Tuple2<Integer, Iterable<WindowedValue<OutputT>>>> combinedDataset = groupedDataset
[beam] 34/45: Fix javadoc of AggregatorCombiner
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 721b1d0ff63222567b8c16484bbf57c693385ffa
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 11:27:28 2019 +0200
Fix javadoc of AggregatorCombiner
---
.../structuredstreaming/translation/batch/AggregatorCombiner.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
index c61d937..138e5c2 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
@@ -41,8 +41,7 @@ import org.apache.spark.sql.expressions.Aggregator;
import org.joda.time.Instant;
import scala.Tuple2;
-/** An {@link Aggregator} for the Spark Batch Runner. It does not use ReduceFnRunner
- * for windowMerging, because reduceFnRunner is based on state which requires a keyed collection.
+/** An {@link Aggregator} for the Spark Batch Runner.
* The accumulator is a {@code Iterable<WindowedValue<AccumT>> because an {@code InputT} can be in multiple windows. So, when accumulating {@code InputT} values, we create one accumulator per input window.
* */
[beam] 22/45: Fix encoder in combine call
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 28067b2b27aeb8a2803216377ee3e296baec2fdb
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon Jun 17 11:53:37 2019 +0200
Fix encoder in combine call
---
.../translation/batch/CombineGloballyTranslatorBatch.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
index fb9e1dd..f29b2c5 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
@@ -59,7 +59,7 @@ class CombineGloballyTranslatorBatch<InputT, AccumT, OutputT>
Dataset<Iterable<WindowedValue<OutputT>>> accumulatedDataset =
combinedRowDataset.map(
- RowHelpers.extractObjectFromRowMapFunction(), EncoderHelpers.windowedValueEncoder());
+ RowHelpers.extractObjectFromRowMapFunction(), EncoderHelpers.genericEncoder());
Dataset<WindowedValue<OutputT>> outputDataset = accumulatedDataset.flatMap(
(FlatMapFunction<Iterable<WindowedValue<OutputT>>, WindowedValue<OutputT>>)
windowedValues -> windowedValues.iterator(), EncoderHelpers.windowedValueEncoder());
[beam] 07/45: Consider null object case on RowHelpers,
fixes empty side inputs tests.
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 9c3b00904ea834d586ad17df53677a3ddc480bd9
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Fri May 10 12:18:08 2019 +0200
Consider null object case on RowHelpers, fixes empty side inputs tests.
---
.../spark/structuredstreaming/translation/helpers/RowHelpers.java | 3 +++
1 file changed, 3 insertions(+)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
index ca88abe..da5cc96 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
@@ -93,6 +93,9 @@ public final class RowHelpers {
public static <T> T extractObjectFromRow(Row value) {
// there is only one value put in each Row by the InputPartitionReader
byte[] bytes = (byte[]) value.get(0);
+ if (bytes == null) {
+ return null;
+ }
ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
Kryo kryo = new Kryo();
Input input = new Input(inputStream);
[beam] 35/45: Fixed immutable list bug
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 78270bfa492c7e08b0615322f51b3c8b02882681
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 15:04:51 2019 +0200
Fixed immutable list bug
---
.../structuredstreaming/translation/batch/AggregatorCombiner.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
index 138e5c2..e005c6f 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombiner.java
@@ -36,6 +36,7 @@ import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Lists;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.expressions.Aggregator;
import org.joda.time.Instant;
@@ -137,7 +138,7 @@ class AggregatorCombiner<K, InputT, AccumT, OutputT, W extends BoundedWindow>
mergedWindowForAccumulator = (mergedWindowForAccumulator == null) ? (W)accumulatorWindow : mergedWindowForAccumulator;
if (mergedWindowToAccumulators.get(mergedWindowForAccumulator) == null){
- mergedWindowToAccumulators.put(mergedWindowForAccumulator, Collections.singletonList(accumulator));
+ mergedWindowToAccumulators.put(mergedWindowForAccumulator, Lists.newArrayList(accumulator));
}
else {
mergedWindowToAccumulators.get(mergedWindowForAccumulator).add(accumulator);
[beam] 26/45: Fix wrong encoder in combineGlobally GBK
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit ddce346faed27c70a3ca0404e1a88248f7dddc38
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jun 27 17:00:19 2019 +0200
Fix wrong encoder in combineGlobally GBK
---
.../translation/batch/CombineGloballyTranslatorBatch.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
index 9219e3f..1a77fc3 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
@@ -81,7 +81,7 @@ class CombineGloballyTranslatorBatch<InputT, AccumT, OutputT>
KeyValueGroupedDataset<Integer, Tuple2<Integer, WindowedValue<InputT>>> groupedDataset = keyedDataset
.groupByKey(
(MapFunction<Tuple2<Integer, WindowedValue<InputT>>, Integer>) value -> value._1(),
- EncoderHelpers.windowedValueEncoder());
+ EncoderHelpers.genericEncoder());
Dataset<Tuple2<Integer, Iterable<WindowedValue<OutputT>>>> combinedDataset = groupedDataset
.agg(new AggregatorCombinerGlobally<>(combineFn, windowingStrategy).toColumn());
[beam] 16/45: Add TODO in Combine translations
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 1a85496dd533364f281ec15cf9aa7f8391405823
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue May 28 16:57:47 2019 +0200
Add TODO in Combine translations
---
.../translation/batch/CombineGloballyTranslatorBatch.java | 1 +
.../translation/batch/CombinePerKeyTranslatorBatch.java | 2 ++
2 files changed, 3 insertions(+)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
index fd66002..53651cf 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
@@ -53,6 +53,7 @@ class CombineGloballyTranslatorBatch<InputT, AccumT, OutputT>
Dataset<WindowedValue<InputT>> inputDataset = context.getDataset(input);
+ //TODO merge windows instead of doing unwindow/window to comply with beam model
Dataset<InputT> unWindowedDataset =
inputDataset.map(WindowingHelpers.unwindowMapFunction(), EncoderHelpers.genericEncoder());
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
index 5ff0048..3d0ee8b 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
@@ -52,9 +52,11 @@ class CombinePerKeyTranslatorBatch<K, InputT, AccumT, OutputT>
Dataset<WindowedValue<KV<K, InputT>>> inputDataset = context.getDataset(input);
+ //TODO merge windows instead of doing unwindow/window to comply with beam model
Dataset<KV<K, InputT>> keyedDataset =
inputDataset.map(WindowingHelpers.unwindowMapFunction(), EncoderHelpers.kvEncoder());
+ // TODO change extractKey impl to deal with WindowedVAlue and use it in GBK
KeyValueGroupedDataset<K, KV<K, InputT>> groupedDataset =
keyedDataset.groupByKey(KVHelpers.extractKey(), EncoderHelpers.genericEncoder());
[beam] 37/45: Clean groupByKeyTest
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit bea2c60e124e4858657ff30c9d67a08fd7ebddb6
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 15:08:12 2019 +0200
Clean groupByKeyTest
---
.../spark/structuredstreaming/translation/batch/GroupByKeyTest.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTest.java
index c95d07d..6e20164 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTest.java
@@ -59,7 +59,7 @@ public class GroupByKeyTest implements Serializable {
@Test
public void testGroupByKeyPreservesWindowing(){
- PCollection<KV<Integer, Iterable<Integer>>> input = p.apply(Create
+ p.apply(Create
.timestamped(
TimestampedValue.of(KV.of(1, 1), new Instant(1)),
TimestampedValue.of(KV.of(1, 3), new Instant(2)),
@@ -90,8 +90,8 @@ public class GroupByKeyTest implements Serializable {
}
}));
p.run();
-
}
+
@Test
public void testGroupByKey() {
List<KV<Integer, Integer>> elems = new ArrayList<>();
[beam] 17/45: Update KVHelpers.extractKey() to deal with
WindowedValue and update GBK and CPK
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 05cb7b8fb0bbf3cfa273209cd4237f2d8d4adecf
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed May 29 11:29:42 2019 +0200
Update KVHelpers.extractKey() to deal with WindowedValue and update GBK and CPK
---
.../translation/batch/CombinePerKeyTranslatorBatch.java | 9 ++-------
.../translation/batch/GroupByKeyTranslatorBatch.java | 4 ++--
.../spark/structuredstreaming/translation/helpers/KVHelpers.java | 5 +++--
3 files changed, 7 insertions(+), 11 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
index 3d0ee8b..c55219b 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
@@ -52,13 +52,8 @@ class CombinePerKeyTranslatorBatch<K, InputT, AccumT, OutputT>
Dataset<WindowedValue<KV<K, InputT>>> inputDataset = context.getDataset(input);
- //TODO merge windows instead of doing unwindow/window to comply with beam model
- Dataset<KV<K, InputT>> keyedDataset =
- inputDataset.map(WindowingHelpers.unwindowMapFunction(), EncoderHelpers.kvEncoder());
-
- // TODO change extractKey impl to deal with WindowedVAlue and use it in GBK
- KeyValueGroupedDataset<K, KV<K, InputT>> groupedDataset =
- keyedDataset.groupByKey(KVHelpers.extractKey(), EncoderHelpers.genericEncoder());
+ KeyValueGroupedDataset<K, WindowedValue<KV<K, InputT>>> groupedDataset =
+ inputDataset.groupByKey(KVHelpers.extractKey(), EncoderHelpers.genericEncoder());
Dataset<Tuple2<K, OutputT>> combinedDataset =
groupedDataset.agg(
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java
index b2b4441..148e643 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java
@@ -28,6 +28,7 @@ import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTr
import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.GroupAlsoByWindowViaOutputBufferFn;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
+import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.KVHelpers;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
@@ -56,8 +57,7 @@ class GroupByKeyTranslatorBatch<K, V>
//group by key only
KeyValueGroupedDataset<K, WindowedValue<KV<K, V>>> groupByKeyOnly = input
- .groupByKey((MapFunction<WindowedValue<KV<K, V>>, K>) wv -> wv.getValue().getKey(),
- EncoderHelpers.genericEncoder());
+ .groupByKey(KVHelpers.extractKey(), EncoderHelpers.genericEncoder());
// Materialize groupByKeyOnly values, potential OOM because of creation of new iterable
Dataset<KV<K, Iterable<WindowedValue<V>>>> materialized = groupByKeyOnly.mapGroups(
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/KVHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/KVHelpers.java
index 3bea466..a53a93f 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/KVHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/KVHelpers.java
@@ -17,6 +17,7 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.helpers;
+import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.spark.api.java.function.MapFunction;
import scala.Tuple2;
@@ -25,8 +26,8 @@ import scala.Tuple2;
public final class KVHelpers {
/** A Spark {@link MapFunction} for extracting the key out of a {@link KV} for GBK for example. */
- public static <K, V> MapFunction<KV<K, V>, K> extractKey() {
- return (MapFunction<KV<K, V>, K>) KV::getKey;
+ public static <K, V> MapFunction<WindowedValue<KV<K, V>>, K> extractKey() {
+ return (MapFunction<WindowedValue<KV<K, V>>, K>) wv -> wv.getValue().getKey();
}
/** A Spark {@link MapFunction} for making a KV out of a {@link scala.Tuple2}. */
[beam] 24/45: Apply a groupByKey avoids for some reason that the
spark structured streaming fmwk casts data to Row which makes it impossible
to deserialize without the coder shipped into the data. For performance
reasons (avoid memory consumption and having to deserialize),
we do not ship coder + data. Also add a mapparitions before GBK to avoid
shuffling
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit e77e95112ab59da871a8c082f9654747fcccb3a9
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed Jun 26 17:44:59 2019 +0200
Apply a groupByKey avoids for some reason that the spark structured streaming fmwk casts data to Row which makes it impossible to deserialize without the coder shipped into the data. For performance reasons (avoid memory consumption and having to deserialize), we do not ship coder + data. Also add a mapparitions before GBK to avoid shuffling
---
.../batch/AggregatorCombinerGlobally.java | 5 ++-
.../batch/CombineGloballyTranslatorBatch.java | 44 ++++++++++++++++++----
2 files changed, 40 insertions(+), 9 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
index d3ad62c..c516629 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
@@ -46,7 +46,7 @@ import scala.Tuple2;
* */
class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindow>
- extends Aggregator<WindowedValue<InputT>, Iterable<WindowedValue<AccumT>>, Iterable<WindowedValue<OutputT>>> {
+ extends Aggregator<Tuple2<Integer, WindowedValue<InputT>>, Iterable<WindowedValue<AccumT>>, Iterable<WindowedValue<OutputT>>> {
private final Combine.CombineFn<InputT, AccumT, OutputT> combineFn;
private WindowingStrategy<InputT, W> windowingStrategy;
@@ -63,8 +63,9 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
}
@Override public Iterable<WindowedValue<AccumT>> reduce(Iterable<WindowedValue<AccumT>> accumulators,
- WindowedValue<InputT> input) {
+ Tuple2<Integer, WindowedValue<InputT>> inputTupe) {
+ WindowedValue<InputT> input = inputTupe._2;
//concatenate accumulators windows and input windows and merge the windows
Collection<W> inputWindows = (Collection<W>)input.getWindows();
Set<W> windows = collectAccumulatorsWindows(accumulators);
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
index f29b2c5..9219e3f 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
@@ -17,18 +17,23 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
-import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.RowHelpers;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.api.java.function.MapPartitionsFunction;
import org.apache.spark.sql.Dataset;
-import org.apache.spark.sql.Row;
+import org.apache.spark.sql.KeyValueGroupedDataset;
+import scala.Tuple2;
/**
* By default Combine.globally is translated as a composite transform that does a Pardo (to key the
@@ -54,12 +59,37 @@ class CombineGloballyTranslatorBatch<InputT, AccumT, OutputT>
WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
Dataset<WindowedValue<InputT>> inputDataset = context.getDataset(input);
- Dataset<Row> combinedRowDataset =
- inputDataset.agg(new AggregatorCombinerGlobally<>(combineFn, windowingStrategy).toColumn());
+ // applying a groupByKey avoids for some reason that the spark structured streaming fmwk
+ // casts data to Row which makes it impossible to deserialize without
+ // the coder shipped into the data. For performance reasons
+ // (avoid memory consumption and having to deserialize), we do not ship coder + data.
+
+ // We do not want to shuffle data during groupByKey, we cannot get the number of partitions for
+ // the input dataset without triggering a costly operation (conversion to rdd) so we cannot use spark Hashpartitioner
+ // so we apply a key to each input dataset partition and then trigger a GBK that should not shuffle data.
+
+ Dataset<Tuple2<Integer, WindowedValue<InputT>>> keyedDataset = inputDataset
+ .mapPartitions((MapPartitionsFunction<WindowedValue<InputT>, Tuple2<Integer, WindowedValue<InputT>>>) inputTIterator -> {
+ List<Tuple2<Integer, WindowedValue<InputT>>> result = new ArrayList<>();
+ Random random = new Random();
+ while (inputTIterator.hasNext()) {
+ result.add(Tuple2.apply(random.nextInt(), inputTIterator.next()));
+ }
+ return result.iterator();
+ }, EncoderHelpers.tuple2Encoder());
+
+ KeyValueGroupedDataset<Integer, Tuple2<Integer, WindowedValue<InputT>>> groupedDataset = keyedDataset
+ .groupByKey(
+ (MapFunction<Tuple2<Integer, WindowedValue<InputT>>, Integer>) value -> value._1(),
+ EncoderHelpers.windowedValueEncoder());
+
+ Dataset<Tuple2<Integer, Iterable<WindowedValue<OutputT>>>> combinedDataset = groupedDataset
+ .agg(new AggregatorCombinerGlobally<>(combineFn, windowingStrategy).toColumn());
+
+ Dataset<Iterable<WindowedValue<OutputT>>> accumulatedDataset = combinedDataset.map(
+ (MapFunction<Tuple2<Integer, Iterable<WindowedValue<OutputT>>>, Iterable<WindowedValue<OutputT>>>) value -> value._2,
+ EncoderHelpers.genericEncoder());
- Dataset<Iterable<WindowedValue<OutputT>>> accumulatedDataset =
- combinedRowDataset.map(
- RowHelpers.extractObjectFromRowMapFunction(), EncoderHelpers.genericEncoder());
Dataset<WindowedValue<OutputT>> outputDataset = accumulatedDataset.flatMap(
(FlatMapFunction<Iterable<WindowedValue<OutputT>>, WindowedValue<OutputT>>)
windowedValues -> windowedValues.iterator(), EncoderHelpers.windowedValueEncoder());
[beam] 29/45: Remove CombineGlobally translation because it is less
performant than the beam sdk one (key + combinePerKey.withHotkeyFanout)
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 18a2e2a6cea9d32c3a0e4495a3dc605e51b21d6d
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed Jul 3 16:57:55 2019 +0200
Remove CombineGlobally translation because it is less performant than the beam sdk one (key + combinePerKey.withHotkeyFanout)
---
.../batch/CombineGloballyTranslatorBatch.java | 83 ----------------------
.../translation/batch/PipelineTranslatorBatch.java | 1 -
2 files changed, 84 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
deleted file mode 100644
index 30101dc..0000000
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.spark.structuredstreaming.translation.batch;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Random;
-import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
-import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
-import org.apache.beam.sdk.transforms.Combine;
-import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.util.WindowedValue;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.WindowingStrategy;
-import org.apache.spark.api.java.function.FlatMapFunction;
-import org.apache.spark.api.java.function.MapFunction;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
-import org.apache.spark.sql.Dataset;
-import org.apache.spark.sql.KeyValueGroupedDataset;
-import scala.Tuple2;
-
-/**
- * By default Combine.globally is translated as a composite transform that does a Pardo (to key the
- * input PCollection with Void keys and then a Combine.PerKey transform. The problem is that
- * Combine.PerKey uses spark GroupByKey which is not very performant due to shuffle. So we add a
- * special CombineGloballyTranslator that does not need GroupByKey
- */
-class CombineGloballyTranslatorBatch<InputT, AccumT, OutputT>
- implements TransformTranslator<PTransform<PCollection<InputT>, PCollection<OutputT>>> {
-
- @Override
- public void translateTransform(
- PTransform<PCollection<InputT>, PCollection<OutputT>> transform, TranslationContext context) {
-
- Combine.Globally combineTransform = (Combine.Globally) transform;
- @SuppressWarnings("unchecked")
- final PCollection<InputT> input = (PCollection<InputT>) context.getInput();
- @SuppressWarnings("unchecked")
- final PCollection<OutputT> output = (PCollection<OutputT>) context.getOutput();
- @SuppressWarnings("unchecked")
- final Combine.CombineFn<InputT, AccumT, OutputT> combineFn =
- (Combine.CombineFn<InputT, AccumT, OutputT>) combineTransform.getFn();
- WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
- Dataset<WindowedValue<InputT>> inputDataset = context.getDataset(input);
-
- // applying a groupByKey avoids for some reason that the spark structured streaming fmwk
- // casts data to Row which makes it impossible to deserialize without
- // the coder shipped into the data. For performance reasons
- // (avoid memory consumption and having to deserialize), we do not ship coder + data.
-
- KeyValueGroupedDataset<Integer, WindowedValue<InputT>> groupedDataset = inputDataset
- .groupByKey((MapFunction<WindowedValue<InputT>, Integer>) value -> 1,
- EncoderHelpers.genericEncoder());
-
- Dataset<Tuple2<Integer, Iterable<WindowedValue<OutputT>>>> combinedDataset = groupedDataset
- .agg(new AggregatorCombinerGlobally<>(combineFn, windowingStrategy).toColumn());
-
- Dataset<Iterable<WindowedValue<OutputT>>> accumulatedDataset = combinedDataset.map(
- (MapFunction<Tuple2<Integer, Iterable<WindowedValue<OutputT>>>, Iterable<WindowedValue<OutputT>>>) value -> value._2,
- EncoderHelpers.genericEncoder());
-
- Dataset<WindowedValue<OutputT>> outputDataset = accumulatedDataset.flatMap(
- (FlatMapFunction<Iterable<WindowedValue<OutputT>>, WindowedValue<OutputT>>)
- windowedValues -> windowedValues.iterator(), EncoderHelpers.windowedValueEncoder());
- context.putDataset(output, outputDataset);
- }
-}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/PipelineTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/PipelineTranslatorBatch.java
index 7f7d962..83a9491 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/PipelineTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/PipelineTranslatorBatch.java
@@ -60,7 +60,6 @@ public class PipelineTranslatorBatch extends PipelineTranslator {
static {
TRANSFORM_TRANSLATORS.put(Combine.PerKey.class, new CombinePerKeyTranslatorBatch());
- TRANSFORM_TRANSLATORS.put(Combine.Globally.class, new CombineGloballyTranslatorBatch());
TRANSFORM_TRANSLATORS.put(GroupByKey.class, new GroupByKeyTranslatorBatch());
// TODO: Do we need to have a dedicated translator for {@code Reshuffle} if it's deprecated?
[beam] 01/45: Build fixup: enable batch Validates Runner tests for
Structured Streaming Runner
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 82e28d1ea8c09853d1fed2a4bdd98f73b61d15c7
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Fri May 3 17:26:26 2019 +0200
Build fixup: enable batch Validates Runner tests for Structured Streaming Runner
---
runners/spark/build.gradle | 49 ++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 49 insertions(+)
diff --git a/runners/spark/build.gradle b/runners/spark/build.gradle
index 8aee17d..db6d3a0 100644
--- a/runners/spark/build.gradle
+++ b/runners/spark/build.gradle
@@ -165,11 +165,60 @@ task validatesRunnerStreaming(type: Test) {
}
}
+task validatesStructuredStreamingRunnerBatch(type: Test) {
+ group = "Verification"
+ def pipelineOptions = JsonOutput.toJson([
+ "--runner=SparkStructuredStreamingRunner",
+ "--streaming=false",
+ ])
+ systemProperty "beamTestPipelineOptions", pipelineOptions
+ systemProperty "beam.spark.test.reuseSparkContext", "true"
+ systemProperty "spark.ui.enabled", "false"
+ systemProperty "spark.ui.showConsoleProgress", "false"
+
+ classpath = configurations.validatesRunner
+ testClassesDirs = files(project(":beam-sdks-java-core").sourceSets.test.output.classesDirs) + files(project(":beam-sdks-java-io-hadoop-format").sourceSets.test.output.classesDirs) + files(project.sourceSets.test.output.classesDirs)
+ // Only one SparkContext may be running in a JVM (SPARK-2243)
+ forkEvery 1
+ maxParallelForks 4
+ useJUnit {
+ includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesCustomWindowMerging'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesFailureMessage'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesParDoLifecycle'
+ // Unbounded
+ excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedPCollections'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream'
+ // State and Timers
+ excludeCategories 'org.apache.beam.sdk.testing.UsesStatefulParDo'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesMapState'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesSetState'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesTimersInParDo'
+ // Metrics
+ excludeCategories 'org.apache.beam.sdk.testing.UsesAttemptedMetrics'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesCommittedMetrics'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesCounterMetrics'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesDistributionMetrics'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesGaugeMetrics'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesMetricsPusher'
+ // SDF
+ excludeCategories 'org.apache.beam.sdk.testing.UsesBoundedSplittableParDo'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesSplittableParDoWithWindowedSideInputs'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedSplittableParDo'
+ // Portability
+ excludeCategories 'org.apache.beam.sdk.testing.UsesImpulse'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesCrossLanguageTransforms'
+ // Schema
+ excludeCategories 'org.apache.beam.sdk.testing.UsesSchema'
+ }
+}
+
task validatesRunner {
group = "Verification"
description "Validates Spark runner"
dependsOn validatesRunnerBatch
dependsOn validatesRunnerStreaming
+ dependsOn validatesStructuredStreamingRunnerBatch
}
// Generates :beam-runners-spark:runQuickstartJavaSpark
[beam] 39/45: Ignore for now not working test testCombineGlobally
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 703e773cc68da44c54864bda116ade723bf369b0
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Fri Jul 5 15:51:54 2019 +0200
Ignore for now not working test testCombineGlobally
---
.../spark/structuredstreaming/translation/batch/CombineTest.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
index f48f79e..ef4468b 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
@@ -35,6 +35,7 @@ import org.apache.beam.sdk.values.TimestampedValue;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.BeforeClass;
+import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@@ -51,6 +52,7 @@ public class CombineTest implements Serializable {
p = Pipeline.create(options);
}
+ @Ignore
@Test
public void testCombineGlobally() {
PCollection<Integer> input =
[beam] 42/45: Add setEnableSparkMetricSinks() method
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a3d1061899d539d84cef97cd3b3f7b2652a04cf7
Author: Alexey Romanenko <ar...@gmail.com>
AuthorDate: Mon Jul 8 18:02:34 2019 +0200
Add setEnableSparkMetricSinks() method
---
.../structuredstreaming/SparkStructuredStreamingPipelineOptions.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
index 27743bd..bbf89f6 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
@@ -30,4 +30,6 @@ public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipe
@Description("Enable/disable sending aggregator values to Spark's metric sinks")
@Default.Boolean(true)
Boolean getEnableSparkMetricSinks();
+
+ void setEnableSparkMetricSinks(Boolean enableSparkMetricSinks);
}
[beam] 08/45: Put back batch/simpleSourceTest.testBoundedSource
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 9d29d76da76c94fffd42557e2d871c86ea809e1a
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue Apr 30 15:10:09 2019 +0200
Put back batch/simpleSourceTest.testBoundedSource
---
.../translation/batch/SimpleSourceTest.java | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/SimpleSourceTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/SimpleSourceTest.java
index 8bd5b24..51be8e3 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/SimpleSourceTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/SimpleSourceTest.java
@@ -31,6 +31,9 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.BoundedSource;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
import org.apache.spark.sql.sources.v2.DataSourceOptions;
import org.apache.spark.sql.sources.v2.reader.DataSourceReader;
import org.junit.BeforeClass;
@@ -85,4 +88,10 @@ public class SimpleSourceTest implements Serializable {
new DatasetSourceBatch().createReader(new DataSourceOptions(dataSourceOptions));
SerializationDebugger.testSerialization(objectToTest, TEMPORARY_FOLDER.newFile());
}
-}
+
+ @Test
+ public void testBoundedSource() {
+ PCollection<Integer> input = p.apply(Create.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
+ PAssert.that(input).containsInAnyOrder(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+ p.run();
+ }}
[beam] 19/45: Implement reduce part of CombineGlobally translation
with windowing
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 233e53206c3675934316d38d0a9424070255e152
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jun 13 11:23:52 2019 +0200
Implement reduce part of CombineGlobally translation with windowing
---
.../batch/AggregatorCombinerGlobally.java | 165 +++++++++++++++++----
.../batch/CombineGloballyTranslatorBatch.java | 19 +--
2 files changed, 144 insertions(+), 40 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
index 2f8293b..0d13218 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
@@ -18,60 +18,173 @@
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
-import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.RowHelpers;
import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.spark.sql.Encoder;
-import org.apache.spark.sql.Row;
import org.apache.spark.sql.expressions.Aggregator;
+import org.joda.time.Instant;
+import scala.Tuple2;
-/** An {@link Aggregator} for the Spark Batch Runner. */
-class AggregatorCombinerGlobally<InputT, AccumT, OutputT>
- extends Aggregator<InputT, AccumT, OutputT> {
+/** An {@link Aggregator} for the Spark Batch Runner. It does not use ReduceFnRunner
+ * for windowMerging, because reduceFnRunner is based on state which requires a keyed collection.
+ * The accumulator is a {@code Iterable<WindowedValue<AccumT>> because an {@code InputT} can be in multiple windows. So, when accumulating {@code InputT} values, we create one accumulator per input window.
+ * */
+
+class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindow>
+ extends Aggregator<WindowedValue<InputT>, Iterable<WindowedValue<AccumT>>, WindowedValue<OutputT>> {
private final Combine.CombineFn<InputT, AccumT, OutputT> combineFn;
+ private WindowingStrategy<InputT, W> windowingStrategy;
+ private TimestampCombiner timestampCombiner;
- public AggregatorCombinerGlobally(Combine.CombineFn<InputT, AccumT, OutputT> combineFn) {
+ public AggregatorCombinerGlobally(Combine.CombineFn<InputT, AccumT, OutputT> combineFn, WindowingStrategy<?, ?> windowingStrategy) {
this.combineFn = combineFn;
+ this.windowingStrategy = (WindowingStrategy<InputT, W>) windowingStrategy;
+ this.timestampCombiner = windowingStrategy.getTimestampCombiner();
}
- @Override
- public AccumT zero() {
- return combineFn.createAccumulator();
+ @Override public Iterable<WindowedValue<AccumT>> zero() {
+ return new ArrayList<>();
}
- @Override
- public AccumT reduce(AccumT accumulator, InputT input) {
- // because of generic type InputT, spark cannot infer an input type.
- // it would pass Integer as input if we had a Aggregator<Integer, ..., ...>
- // without the type inference it stores input in a GenericRowWithSchema
- Row row = (Row) input;
- InputT t = RowHelpers.extractObjectFromRow(row);
- return combineFn.addInput(accumulator, t);
+ @Override public Iterable<WindowedValue<AccumT>> reduce(Iterable<WindowedValue<AccumT>> accumulators,
+ WindowedValue<InputT> input) {
+
+ //concatenate accumulators windows and input windows and merge the windows
+ Collection<W> inputWindows = (Collection<W>)input.getWindows();
+ Set<W> windows = collectAccumulatorsWindows(accumulators);
+ windows.addAll(inputWindows);
+ Map<W, W> windowToMergeResult = null;
+ try {
+ windowToMergeResult = mergeWindows(windowingStrategy, windows);
+ } catch (Exception e) {
+ throw new RuntimeException("Unable to merge accumulators windows and input windows", e);
+ }
+
+ // iterate through the input windows and for each, create an accumulator with the merged window
+ // associated to it and call addInput with the accumulator.
+ // Maintain a map of the accumulators for use as output
+ Map<W, Tuple2<AccumT, Instant>> mapState = new HashMap<>();
+ for (W inputWindow:inputWindows) {
+ W mergedWindow = windowToMergeResult.get(inputWindow);
+ mergedWindow = mergedWindow == null ? inputWindow : mergedWindow;
+ Tuple2<AccumT, Instant> accumAndInstant = mapState.get(mergedWindow);
+ // if there is no accumulator associated with this window yet, create one
+ if (accumAndInstant == null) {
+ AccumT accum = combineFn.addInput(combineFn.createAccumulator(), input.getValue());
+ Instant windowTimestamp =
+ timestampCombiner.assign(
+ mergedWindow, windowingStrategy.getWindowFn().getOutputTime(input.getTimestamp(), mergedWindow));
+ accumAndInstant = new Tuple2<>(accum, windowTimestamp);
+ mapState.put(mergedWindow, accumAndInstant);
+ } else {
+ AccumT updatedAccum =
+ combineFn.addInput(accumAndInstant._1, input.getValue());
+ Instant updatedTimestamp = timestampCombiner.combine(accumAndInstant._2, timestampCombiner
+ .assign(mergedWindow,
+ windowingStrategy.getWindowFn().getOutputTime(input.getTimestamp(), mergedWindow)));
+ accumAndInstant = new Tuple2<>(updatedAccum, updatedTimestamp);
+ }
+ }
+ // output the accumulators map
+ List<WindowedValue<AccumT>> result = new ArrayList<>();
+ for (Map.Entry<W, Tuple2<AccumT, Instant>> entry : mapState.entrySet()) {
+ AccumT accumulator = entry.getValue()._1;
+ Instant windowTimestamp = entry.getValue()._2;
+ W window = entry.getKey();
+ result.add(WindowedValue.of(accumulator, windowTimestamp, window, PaneInfo.NO_FIRING));
+ }
+ return result;
}
- @Override
- public AccumT merge(AccumT accumulator1, AccumT accumulator2) {
+ @Override public Iterable<WindowedValue<AccumT>> merge(
+ Iterable<WindowedValue<AccumT>> accumulators1,
+ Iterable<WindowedValue<AccumT>> accumulators2) {
+ // TODO
+ /*
ArrayList<AccumT> accumulators = new ArrayList<>();
accumulators.add(accumulator1);
accumulators.add(accumulator2);
return combineFn.mergeAccumulators(accumulators);
+*/
+ return null;
}
- @Override
- public OutputT finish(AccumT reduction) {
- return combineFn.extractOutput(reduction);
+ @Override public WindowedValue<OutputT> finish(Iterable<WindowedValue<AccumT>> reduction) {
+ // TODO
+ // return combineFn.extractOutput(reduction);
+ return null;
}
- @Override
- public Encoder<AccumT> bufferEncoder() {
+ @Override public Encoder<Iterable<WindowedValue<AccumT>>> bufferEncoder() {
// TODO replace with accumulatorCoder if possible
return EncoderHelpers.genericEncoder();
}
- @Override
- public Encoder<OutputT> outputEncoder() {
+ @Override public Encoder<WindowedValue<OutputT>> outputEncoder() {
// TODO replace with outputCoder if possible
return EncoderHelpers.genericEncoder();
}
+
+ private Set<W> collectAccumulatorsWindows(Iterable<WindowedValue<AccumT>> accumulators) {
+ Set<W> windows = new HashSet<>();
+ for (WindowedValue<?> accumulator : accumulators) {
+ // an accumulator has only one window associated to it.
+ W accumulatorWindow = (W) accumulator.getWindows().iterator().next();
+ windows.add(accumulatorWindow);
+ } return windows;
+ }
+
+ private Map<W, W> mergeWindows(WindowingStrategy<InputT, W> windowingStrategy, Set<W> windows)
+ throws Exception {
+ WindowFn<InputT, W> windowFn = windowingStrategy.getWindowFn();
+
+ if (windowingStrategy.getWindowFn().isNonMerging()) {
+ // Return an empty map, indicating that every window is not merged.
+ return Collections.emptyMap();
+ }
+
+ Map<W, W> windowToMergeResult = new HashMap<>();
+ windowFn.mergeWindows(new MergeContextImpl(windowFn, windows, windowToMergeResult));
+ return windowToMergeResult;
+ }
+
+
+ private class MergeContextImpl extends WindowFn<InputT, W>.MergeContext {
+
+ private Set<W> windows;
+ private Map<W, W> windowToMergeResult;
+
+ MergeContextImpl(WindowFn<InputT, W> windowFn, Set<W> windows, Map<W, W> windowToMergeResult) {
+ windowFn.super();
+ this.windows = windows;
+ this.windowToMergeResult = windowToMergeResult;
+ }
+
+ @Override
+ public Collection<W> windows() {
+ return windows;
+ }
+
+ @Override
+ public void merge(Collection<W> toBeMerged, W mergeResult) throws Exception {
+ for (W w : toBeMerged) {
+ windowToMergeResult.put(w, mergeResult);
+ }
+ }
+ }
+
}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
index 53651cf..f18572b 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineGloballyTranslatorBatch.java
@@ -26,6 +26,7 @@ import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
@@ -50,25 +51,15 @@ class CombineGloballyTranslatorBatch<InputT, AccumT, OutputT>
@SuppressWarnings("unchecked")
final Combine.CombineFn<InputT, AccumT, OutputT> combineFn =
(Combine.CombineFn<InputT, AccumT, OutputT>) combineTransform.getFn();
-
+ WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
Dataset<WindowedValue<InputT>> inputDataset = context.getDataset(input);
- //TODO merge windows instead of doing unwindow/window to comply with beam model
- Dataset<InputT> unWindowedDataset =
- inputDataset.map(WindowingHelpers.unwindowMapFunction(), EncoderHelpers.genericEncoder());
-
Dataset<Row> combinedRowDataset =
- unWindowedDataset.agg(new AggregatorCombinerGlobally<>(combineFn).toColumn());
-
- Dataset<OutputT> combinedDataset =
- combinedRowDataset.map(
- RowHelpers.extractObjectFromRowMapFunction(), EncoderHelpers.genericEncoder());
+ inputDataset.agg(new AggregatorCombinerGlobally<>(combineFn, windowingStrategy).toColumn());
- // Window the result into global window.
Dataset<WindowedValue<OutputT>> outputDataset =
- combinedDataset.map(
- WindowingHelpers.windowMapFunction(), EncoderHelpers.windowedValueEncoder());
-
+ combinedRowDataset.map(
+ RowHelpers.extractObjectFromRowMapFunction(), EncoderHelpers.windowedValueEncoder());
context.putDataset(output, outputDataset);
}
}
[beam] 02/45: Build fixup: Limit the number of partitions to make
tests go 300% faster
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 1d52f356b218934d0e596423b464d0e916ee45af
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Fri May 3 22:15:36 2019 +0200
Build fixup: Limit the number of partitions to make tests go 300% faster
---
runners/spark/build.gradle | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/runners/spark/build.gradle b/runners/spark/build.gradle
index db6d3a0..72e6013 100644
--- a/runners/spark/build.gradle
+++ b/runners/spark/build.gradle
@@ -37,6 +37,7 @@ configurations {
test {
systemProperty "beam.spark.test.reuseSparkContext", "true"
+ systemProperty "spark.sql.shuffle.partitions", "4"
systemProperty "spark.ui.enabled", "false"
systemProperty "spark.ui.showConsoleProgress", "false"
systemProperty "beamTestPipelineOptions", """[
@@ -172,7 +173,7 @@ task validatesStructuredStreamingRunnerBatch(type: Test) {
"--streaming=false",
])
systemProperty "beamTestPipelineOptions", pipelineOptions
- systemProperty "beam.spark.test.reuseSparkContext", "true"
+ systemProperty "spark.sql.shuffle.partitions", "4"
systemProperty "spark.ui.enabled", "false"
systemProperty "spark.ui.showConsoleProgress", "false"
[beam] 09/45: Update windowAssignTest
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit ab26efca8b2e9be20036a50c6728692e80d8a7d0
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue May 7 09:35:46 2019 +0200
Update windowAssignTest
---
.../translation/batch/WindowAssignTest.java | 19 ++++++++-----------
1 file changed, 8 insertions(+), 11 deletions(-)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/WindowAssignTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/WindowAssignTest.java
index 61da3ea..3011d88 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/WindowAssignTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/WindowAssignTest.java
@@ -27,13 +27,11 @@ import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
-import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TimestampedValue;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@@ -50,20 +48,19 @@ public class WindowAssignTest implements Serializable {
p = Pipeline.create(options);
}
- @Ignore
@Test
public void testWindowAssign() {
- PCollection<KV<Integer, Integer>> input =
+ PCollection<Integer> input =
p.apply(
Create.timestamped(
- TimestampedValue.of(KV.of(1, 1), new Instant(1)),
- TimestampedValue.of(KV.of(1, 2), new Instant(2)),
- TimestampedValue.of(KV.of(1, 3), new Instant(3)),
- TimestampedValue.of(KV.of(1, 4), new Instant(10)),
- TimestampedValue.of(KV.of(1, 5), new Instant(11))))
+ TimestampedValue.of(1, new Instant(1)),
+ TimestampedValue.of(2, new Instant(2)),
+ TimestampedValue.of(3, new Instant(3)),
+ TimestampedValue.of(4, new Instant(10)),
+ TimestampedValue.of(5, new Instant(11))))
.apply(Window.into(FixedWindows.of(Duration.millis(10))))
- .apply(Sum.integersPerKey());
- PAssert.that(input).containsInAnyOrder(KV.of(1, 6), KV.of(1, 9));
+ .apply(Sum.integersGlobally().withoutDefaults());
+ PAssert.that(input).containsInAnyOrder(6, 9);
p.run();
}
}
[beam] 33/45: Clean not more needed WindowingHelpers
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit ea959c6391c2bd44b29e1d19c5a7c0a6cbd4fc53
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 11:25:31 2019 +0200
Clean not more needed WindowingHelpers
---
.../translation/helpers/WindowingHelpers.java | 22 ----------------------
1 file changed, 22 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/WindowingHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/WindowingHelpers.java
index 448918a..38ba54b 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/WindowingHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/WindowingHelpers.java
@@ -33,28 +33,6 @@ import org.joda.time.Instant;
public final class WindowingHelpers {
/**
- * A Spark {@link MapFunction} for converting a value to a {@link WindowedValue}. The resulting
- * {@link WindowedValue} will be in a global windows, and will have the default timestamp == MIN
- * and pane.
- *
- * @param <T> The type of the object.
- * @return A {@link MapFunction} that accepts an object and returns its {@link WindowedValue}.
- */
- public static <T> MapFunction<T, WindowedValue<T>> windowMapFunction() {
- return (MapFunction<T, WindowedValue<T>>) WindowedValue::valueInGlobalWindow;
- }
-
- /**
- * A Spark {@link MapFunction} for extracting the value from a {@link WindowedValue}.
- *
- * @param <T> The type of the object.
- * @return A {@link MapFunction} that accepts a {@link WindowedValue} and returns its value.
- */
- public static <T> MapFunction<WindowedValue<T>, T> unwindowMapFunction() {
- return (MapFunction<WindowedValue<T>, T>) WindowedValue::getValue;
- }
-
- /**
* Checks if the window transformation should be applied or skipped.
*
* <p>Avoid running assign windows if both source and destination are global window or if the user
[beam] 25/45: Fix case when a window does not merge into any other
window
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 3ae7e09ade321873f1c05d2f93cdebe2d2cdf446
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jun 27 16:59:47 2019 +0200
Fix case when a window does not merge into any other window
---
.../translation/batch/AggregatorCombinerGlobally.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
index c516629..884f51e 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
@@ -133,6 +133,8 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
//each accumulator has only one window
BoundedWindow accumulatorWindow = accumulator.getWindows().iterator().next();
W mergedWindowForAccumulator = windowToMergeResult.get(accumulatorWindow);
+ mergedWindowForAccumulator = (mergedWindowForAccumulator == null) ? (W)accumulatorWindow : mergedWindowForAccumulator;
+
if (mergedWindowToAccumulators.get(mergedWindowForAccumulator) == null){
mergedWindowToAccumulators.put(mergedWindowForAccumulator, Collections.singletonList(accumulator));
}
[beam] 21/45: Implement merge accumulators part of CombineGlobally
translation with windowing
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a4af0317f3a9b766df197ecd505a8b77342e3017
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Sun Jun 16 12:12:09 2019 +0200
Implement merge accumulators part of CombineGlobally translation with windowing
---
.../batch/AggregatorCombinerGlobally.java | 48 ++++++++++++++++++----
1 file changed, 39 insertions(+), 9 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
index 6996165..d3ad62c 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
@@ -25,6 +25,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -33,6 +34,7 @@ import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.WindowingStrategy;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.expressions.Aggregator;
import org.joda.time.Instant;
@@ -67,7 +69,7 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
Collection<W> inputWindows = (Collection<W>)input.getWindows();
Set<W> windows = collectAccumulatorsWindows(accumulators);
windows.addAll(inputWindows);
- Map<W, W> windowToMergeResult = null;
+ Map<W, W> windowToMergeResult;
try {
windowToMergeResult = mergeWindows(windowingStrategy, windows);
} catch (Exception e) {
@@ -113,14 +115,42 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
@Override public Iterable<WindowedValue<AccumT>> merge(
Iterable<WindowedValue<AccumT>> accumulators1,
Iterable<WindowedValue<AccumT>> accumulators2) {
- // TODO
- /*
- ArrayList<AccumT> accumulators = new ArrayList<>();
- accumulators.add(accumulator1);
- accumulators.add(accumulator2);
- return combineFn.mergeAccumulators(accumulators);
-*/
- return null;
+
+ // merge the windows of all the accumulators
+ Iterable<WindowedValue<AccumT>> accumulators = Iterables.concat(accumulators1, accumulators2);
+ Set<W> accumulatorsWindows = collectAccumulatorsWindows(accumulators);
+ Map<W, W> windowToMergeResult;
+ try {
+ windowToMergeResult = mergeWindows(windowingStrategy, accumulatorsWindows);
+ } catch (Exception e) {
+ throw new RuntimeException("Unable to merge accumulators windows", e);
+ }
+
+ // group accumulators by their merged window
+ Map<W, List<WindowedValue<AccumT>>> mergedWindowToAccumulators = new HashMap<>();
+ for (WindowedValue<AccumT> accumulator : accumulators) {
+ //each accumulator has only one window
+ BoundedWindow accumulatorWindow = accumulator.getWindows().iterator().next();
+ W mergedWindowForAccumulator = windowToMergeResult.get(accumulatorWindow);
+ if (mergedWindowToAccumulators.get(mergedWindowForAccumulator) == null){
+ mergedWindowToAccumulators.put(mergedWindowForAccumulator, Collections.singletonList(accumulator));
+ }
+ else {
+ mergedWindowToAccumulators.get(mergedWindowForAccumulator).add(accumulator);
+ }
+ }
+ // merge the accumulators for each mergedWindow
+ List<WindowedValue<AccumT>> result = new ArrayList<>();
+ for (Map.Entry<W, List<WindowedValue<AccumT>>> entry : mergedWindowToAccumulators.entrySet()){
+ W mergedWindow = entry.getKey();
+ List<WindowedValue<AccumT>> accumulatorsForMergedWindow = entry.getValue();
+ result.add(WindowedValue
+ .of(combineFn.mergeAccumulators(accumulatorsForMergedWindow.stream().map(x -> x.getValue()).collect(
+ Collectors.toList())), timestampCombiner.combine(accumulatorsForMergedWindow.stream().map(x -> x.getTimestamp()).collect(
+ Collectors.toList())),
+ mergedWindow, PaneInfo.NO_FIRING));
+ }
+ return result;
}
@Override public Iterable<WindowedValue<OutputT>> finish(Iterable<WindowedValue<AccumT>> reduction) {
[beam] 23/45: Revert extractKey while combinePerKey is not done (so
that it compiles)
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit b1cf53610f0688edd0fc3449a4775ae90d330c43
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon Jun 17 13:59:13 2019 +0200
Revert extractKey while combinePerKey is not done (so that it compiles)
---
.../translation/batch/CombinePerKeyTranslatorBatch.java | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
index c55219b..1c35301 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombinePerKeyTranslatorBatch.java
@@ -28,6 +28,7 @@ import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
+import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.KeyValueGroupedDataset;
import scala.Tuple2;
@@ -52,8 +53,10 @@ class CombinePerKeyTranslatorBatch<K, InputT, AccumT, OutputT>
Dataset<WindowedValue<KV<K, InputT>>> inputDataset = context.getDataset(input);
- KeyValueGroupedDataset<K, WindowedValue<KV<K, InputT>>> groupedDataset =
- inputDataset.groupByKey(KVHelpers.extractKey(), EncoderHelpers.genericEncoder());
+ Dataset<KV<K, InputT>> unwindowedDataset = inputDataset
+ .map(WindowingHelpers.unwindowMapFunction(), EncoderHelpers.kvEncoder());
+ KeyValueGroupedDataset<K, KV<K, InputT>> groupedDataset =
+ unwindowedDataset.groupByKey((MapFunction<KV<K, InputT>, K>) kv -> kv.getKey(), EncoderHelpers.genericEncoder());
Dataset<Tuple2<K, OutputT>> combinedDataset =
groupedDataset.agg(
[beam] 41/45: Add missing dependencies to run Spark Structured
Streaming Runner on Nexmark
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit f88db1b9628a9ac7bbe200205b1df8a0e963d4bb
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Sun Apr 21 11:59:04 2019 +0200
Add missing dependencies to run Spark Structured Streaming Runner on Nexmark
---
sdks/java/testing/nexmark/build.gradle | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/sdks/java/testing/nexmark/build.gradle b/sdks/java/testing/nexmark/build.gradle
index a1db48a..19b39d4 100644
--- a/sdks/java/testing/nexmark/build.gradle
+++ b/sdks/java/testing/nexmark/build.gradle
@@ -82,10 +82,11 @@ dependencies {
// runs with the Spark runner, we can provide such a dependency. This is deliberately phrased
// to not hardcode any runner other than :beam-runners-direct-java
if (shouldProvideSpark) {
- gradleRun library.java.spark_streaming
gradleRun library.java.spark_core, {
exclude group:"org.slf4j", module:"jul-to-slf4j"
}
+ gradleRun library.java.spark_sql
+ gradleRun library.java.spark_streaming
}
}
[beam] 45/45: Add a test to check that CombineGlobally preserves
windowing
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a37817db9b787e1818519bfc330f809be809ce1c
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue Jul 9 14:57:18 2019 +0200
Add a test to check that CombineGlobally preserves windowing
---
.../translation/batch/CombineTest.java | 16 +++++++++++++++-
1 file changed, 15 insertions(+), 1 deletion(-)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
index 3505da5..cae46a6 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
@@ -25,6 +25,7 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.Sum;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
@@ -35,7 +36,6 @@ import org.apache.beam.sdk.values.TimestampedValue;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.BeforeClass;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@@ -62,6 +62,20 @@ public class CombineTest implements Serializable {
}
@Test
+ public void testCombineGloballyPreservesWindowing() {
+ PCollection<Integer> input = p.apply(Create.timestamped(
+ TimestampedValue.of(1, new Instant(1)),
+ TimestampedValue.of(2, new Instant(2)),
+ TimestampedValue.of(3, new Instant(11)),
+ TimestampedValue.of(4, new Instant(3)),
+ TimestampedValue.of(5, new Instant(11)),
+ TimestampedValue.of(6, new Instant(12))))
+ .apply(Window.into(FixedWindows.of(Duration.millis(10))))
+ .apply(Combine.globally(Sum.ofIntegers()).withoutDefaults());
+ PAssert.that(input).containsInAnyOrder(7, 14);
+ }
+
+ @Test
public void testCombinePerKey() {
List<KV<Integer, Integer>> elems = new ArrayList<>();
elems.add(KV.of(1, 1));
[beam] 31/45: Clean no more needed KVHelpers
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 2851c72180d2521dcdc382f7879528d25bcbb98f
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 11:23:26 2019 +0200
Clean no more needed KVHelpers
---
.../spark/structuredstreaming/translation/helpers/KVHelpers.java | 5 -----
1 file changed, 5 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/KVHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/KVHelpers.java
index a53a93f..0e37551 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/KVHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/KVHelpers.java
@@ -29,9 +29,4 @@ public final class KVHelpers {
public static <K, V> MapFunction<WindowedValue<KV<K, V>>, K> extractKey() {
return (MapFunction<WindowedValue<KV<K, V>>, K>) wv -> wv.getValue().getKey();
}
-
- /** A Spark {@link MapFunction} for making a KV out of a {@link scala.Tuple2}. */
- public static <K, V> MapFunction<Tuple2<K, V>, KV<K, V>> tuple2ToKV() {
- return (MapFunction<Tuple2<K, V>, KV<K, V>>) tuple2 -> KV.of(tuple2._1, tuple2._2);
- }
}
[beam] 32/45: Clean not more needed RowHelpers
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit e84c0bbccb9a4077a876f094dfc02e54031343f5
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 11:24:36 2019 +0200
Clean not more needed RowHelpers
---
.../translation/helpers/RowHelpers.java | 28 ----------------------
1 file changed, 28 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
index da5cc96..5d96911 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/RowHelpers.java
@@ -76,32 +76,4 @@ public final class RowHelpers {
}
return InternalRow.apply(asScalaBuffer(list).toList());
}
-
- /**
- * A Spark {@link MapFunction} for extracting an Object from a Row was serialized to bytes using
- * kryo.
- *
- * @param <T> The type of the object.
- * @return A {@link MapFunction} that accepts a {@link Row} and returns its {@link WindowedValue}.
- */
- public static <T> MapFunction<Row, T> extractObjectFromRowMapFunction() {
- return (MapFunction<Row, T>) RowHelpers::extractObjectFromRow;
- }
-
- /** Extracts an Object from a Row was serialized to bytes using kryo. */
- @SuppressWarnings("TypeParameterUnusedInFormals")
- public static <T> T extractObjectFromRow(Row value) {
- // there is only one value put in each Row by the InputPartitionReader
- byte[] bytes = (byte[]) value.get(0);
- if (bytes == null) {
- return null;
- }
- ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
- Kryo kryo = new Kryo();
- Input input = new Input(inputStream);
- @SuppressWarnings("unchecked")
- T object = (T) kryo.readClassAndObject(input);
- input.close();
- return object;
- }
}
[beam] 43/45: Fix javadoc
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 0fc97ee9cf704d17b9b8aeb42375af3ecbabbc13
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon Jul 8 15:47:49 2019 +0200
Fix javadoc
---
.../spark/structuredstreaming/translation/TranslationContext.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
index 411aec3..00bdea8 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
@@ -179,7 +179,7 @@ public class TranslationContext {
// Pipeline methods
// --------------------------------------------------------------------------------------------
- /** Starts a new pipeline. */
+ /** Starts the pipeline. */
public void startPipeline(boolean testMode) {
try {
SparkStructuredStreamingPipelineOptions options =
[beam] 11/45: Re-code GroupByKeyTranslatorBatch to conserve
windowing instead of unwindowing/windowing(GlobalWindow): simplify code,
use ReduceFnRunner to merge the windows
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 6bc9b171ee987739382c8d9acc31144ab14b4a01
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon May 27 15:10:02 2019 +0200
Re-code GroupByKeyTranslatorBatch to conserve windowing instead of unwindowing/windowing(GlobalWindow): simplify code, use ReduceFnRunner to merge the windows
---
.../batch/GroupByKeyTranslatorBatch.java | 74 ++++++---
.../GroupAlsoByWindowViaOutputBufferFn.java | 168 +++++++++++++++++++++
2 files changed, 217 insertions(+), 25 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java
index 48cceee..b2b4441 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTranslatorBatch.java
@@ -17,16 +17,23 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
+import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
+import org.apache.beam.runners.core.InMemoryStateInternals;
+import org.apache.beam.runners.core.StateInternals;
+import org.apache.beam.runners.core.StateInternalsFactory;
+import org.apache.beam.runners.core.SystemReduceFn;
import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
+import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.GroupAlsoByWindowViaOutputBufferFn;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
-import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.WindowingHelpers;
+import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.api.java.function.MapGroupsFunction;
@@ -42,34 +49,51 @@ class GroupByKeyTranslatorBatch<K, V>
PTransform<PCollection<KV<K, V>>, PCollection<KV<K, Iterable<V>>>> transform,
TranslationContext context) {
- Dataset<WindowedValue<KV<K, V>>> input = context.getDataset(context.getInput());
+ @SuppressWarnings("unchecked")
+ final PCollection<KV<K, V>> inputPCollection = (PCollection<KV<K, V>>) context.getInput();
- // Extract key to group by key only.
- KeyValueGroupedDataset<K, KV<K, V>> grouped =
- input
- .map(WindowingHelpers.unwindowMapFunction(), EncoderHelpers.kvEncoder())
- .groupByKey((MapFunction<KV<K, V>, K>) KV::getKey, EncoderHelpers.genericEncoder());
+ Dataset<WindowedValue<KV<K, V>>> input = context.getDataset(inputPCollection);
- // Materialize grouped values, potential OOM because of creation of new iterable
- Dataset<KV<K, Iterable<V>>> materialized =
- grouped.mapGroups(
- (MapGroupsFunction<K, KV<K, V>, KV<K, Iterable<V>>>)
- // TODO: We need to improve this part and avoid creating of new List (potential OOM)
- // (key, iterator) -> KV.of(key, () -> Iterators.transform(iterator, KV::getValue)),
- (key, iterator) -> {
- List<V> values = new ArrayList<>();
- while (iterator.hasNext()) {
- values.add(iterator.next().getValue());
- }
- return KV.of(key, Iterables.unmodifiableIterable(values));
- },
- EncoderHelpers.kvEncoder());
+ //group by key only
+ KeyValueGroupedDataset<K, WindowedValue<KV<K, V>>> groupByKeyOnly = input
+ .groupByKey((MapFunction<WindowedValue<KV<K, V>>, K>) wv -> wv.getValue().getKey(),
+ EncoderHelpers.genericEncoder());
- // Window the result into global window.
- Dataset<WindowedValue<KV<K, Iterable<V>>>> output =
- materialized.map(
- WindowingHelpers.windowMapFunction(), EncoderHelpers.windowedValueEncoder());
+ // Materialize groupByKeyOnly values, potential OOM because of creation of new iterable
+ Dataset<KV<K, Iterable<WindowedValue<V>>>> materialized = groupByKeyOnly.mapGroups(
+ (MapGroupsFunction<K, WindowedValue<KV<K, V>>, KV<K, Iterable<WindowedValue<V>>>>) (key, iterator) -> {
+ List<WindowedValue<V>> values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ WindowedValue<KV<K, V>> next = iterator.next();
+ values.add(WindowedValue
+ .of(next.getValue().getValue(), next.getTimestamp(), next.getWindows(),
+ next.getPane()));
+ }
+ KV<K, Iterable<WindowedValue<V>>> kv = KV.of(key, Iterables.unmodifiableIterable(values));
+ return kv;
+ }, EncoderHelpers.kvEncoder());
+
+ WindowingStrategy<?, ?> windowingStrategy = inputPCollection.getWindowingStrategy();
+ KvCoder<K, V> coder = (KvCoder<K, V>) inputPCollection.getCoder();
+ // group also by windows
+ Dataset<WindowedValue<KV<K, Iterable<V>>>> output = materialized.flatMap(
+ new GroupAlsoByWindowViaOutputBufferFn<>(windowingStrategy,
+ new InMemoryStateInternalsFactory<>(), SystemReduceFn.buffering(coder.getValueCoder()),
+ context.getSerializableOptions()), EncoderHelpers.windowedValueEncoder());
context.putDataset(context.getOutput(), output);
}
+
+ /**
+ * In-memory state internals factory.
+ *
+ * @param <K> State key type.
+ */
+ static class InMemoryStateInternalsFactory<K> implements StateInternalsFactory<K>, Serializable {
+ @Override
+ public StateInternals stateInternalsForKey(K key) {
+ return InMemoryStateInternals.forKey(key);
+ }
+ }
+
}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/GroupAlsoByWindowViaOutputBufferFn.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/GroupAlsoByWindowViaOutputBufferFn.java
new file mode 100644
index 0000000..2fb08f5
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/GroupAlsoByWindowViaOutputBufferFn.java
@@ -0,0 +1,168 @@
+/*
+ * 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.spark.structuredstreaming.translation.batch.functions;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import org.apache.beam.runners.core.InMemoryTimerInternals;
+import org.apache.beam.runners.core.OutputWindowedValue;
+import org.apache.beam.runners.core.ReduceFnRunner;
+import org.apache.beam.runners.core.StateInternals;
+import org.apache.beam.runners.core.StateInternalsFactory;
+import org.apache.beam.runners.core.SystemReduceFn;
+import org.apache.beam.runners.core.UnsupportedSideInputReader;
+import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
+import org.apache.beam.runners.core.construction.TriggerTranslation;
+import org.apache.beam.runners.core.triggers.ExecutableTriggerStateMachine;
+import org.apache.beam.runners.core.triggers.TriggerStateMachines;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.joda.time.Instant;
+
+/** A FlatMap function that groups by windows in batch mode using {@link ReduceFnRunner}. */
+public class GroupAlsoByWindowViaOutputBufferFn<K, InputT, W extends BoundedWindow>
+ implements FlatMapFunction<
+ KV<K, Iterable<WindowedValue<InputT>>>,
+ WindowedValue<KV<K, Iterable<InputT>>>> {
+
+ private final WindowingStrategy<?, W> windowingStrategy;
+ private final StateInternalsFactory<K> stateInternalsFactory;
+ private final SystemReduceFn<K, InputT, Iterable<InputT>, Iterable<InputT>, W> reduceFn;
+ private final SerializablePipelineOptions options;
+
+ public GroupAlsoByWindowViaOutputBufferFn(
+ WindowingStrategy<?, W> windowingStrategy,
+ StateInternalsFactory<K> stateInternalsFactory,
+ SystemReduceFn<K, InputT, Iterable<InputT>, Iterable<InputT>, W> reduceFn,
+ SerializablePipelineOptions options) {
+ this.windowingStrategy = windowingStrategy;
+ this.stateInternalsFactory = stateInternalsFactory;
+ this.reduceFn = reduceFn;
+ this.options = options;
+ }
+
+ @Override
+ public Iterator<WindowedValue<KV<K, Iterable<InputT>>>> call(
+ KV<K, Iterable<WindowedValue<InputT>>> kv) throws Exception {
+ K key = kv.getKey();
+ Iterable<WindowedValue<InputT>> values = kv.getValue();
+
+ // ------ based on GroupAlsoByWindowsViaOutputBufferDoFn ------//
+
+ // Used with Batch, we know that all the data is available for this key. We can't use the
+ // timer manager from the context because it doesn't exist. So we create one and emulate the
+ // watermark, knowing that we have all data and it is in timestamp order.
+ InMemoryTimerInternals timerInternals = new InMemoryTimerInternals();
+ timerInternals.advanceProcessingTime(Instant.now());
+ timerInternals.advanceSynchronizedProcessingTime(Instant.now());
+ StateInternals stateInternals = stateInternalsFactory.stateInternalsForKey(key);
+ GABWOutputWindowedValue<K, InputT> outputter = new GABWOutputWindowedValue<>();
+
+ ReduceFnRunner<K, InputT, Iterable<InputT>, W> reduceFnRunner =
+ new ReduceFnRunner<>(
+ key,
+ windowingStrategy,
+ ExecutableTriggerStateMachine.create(
+ TriggerStateMachines.stateMachineForTrigger(
+ TriggerTranslation.toProto(windowingStrategy.getTrigger()))),
+ stateInternals,
+ timerInternals,
+ outputter,
+ new UnsupportedSideInputReader("GroupAlsoByWindow"),
+ reduceFn,
+ options.get());
+
+ // Process the grouped values.
+ reduceFnRunner.processElements(values);
+
+ // Finish any pending windows by advancing the input watermark to infinity.
+ timerInternals.advanceInputWatermark(BoundedWindow.TIMESTAMP_MAX_VALUE);
+
+ // not supported yet
+/*
+ // Finally, advance the processing time to infinity to fire any timers.
+ timerInternals.advanceProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
+ timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
+
+ fireEligibleTimers(timerInternals, reduceFnRunner);
+*/
+
+ reduceFnRunner.persist();
+
+ return outputter.getOutputs().iterator();
+ }
+
+/* private void fireEligibleTimers(
+ InMemoryTimerInternals timerInternals,
+ ReduceFnRunner<K, InputT, Iterable<InputT>, W> reduceFnRunner)
+ throws Exception {
+ List<TimerInternals.TimerData> timers = new ArrayList<>();
+ while (true) {
+ TimerInternals.TimerData timer;
+ while ((timer = timerInternals.removeNextEventTimer()) != null) {
+ timers.add(timer);
+ }
+ while ((timer = timerInternals.removeNextProcessingTimer()) != null) {
+ timers.add(timer);
+ }
+ while ((timer = timerInternals.removeNextSynchronizedProcessingTimer()) != null) {
+ timers.add(timer);
+ }
+ if (timers.isEmpty()) {
+ break;
+ }
+ reduceFnRunner.onTimers(timers);
+ timers.clear();
+ }
+ }*/
+
+ private static class GABWOutputWindowedValue<K, V>
+ implements OutputWindowedValue<KV<K, Iterable<V>>> {
+ private final List<WindowedValue<KV<K, Iterable<V>>>> outputs = new ArrayList<>();
+
+ @Override
+ public void outputWindowedValue(
+ KV<K, Iterable<V>> output,
+ Instant timestamp,
+ Collection<? extends BoundedWindow> windows,
+ PaneInfo pane) {
+ outputs.add(WindowedValue.of(output, timestamp, windows, pane));
+ }
+
+ @Override
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
+ Instant timestamp,
+ Collection<? extends BoundedWindow> windows,
+ PaneInfo pane) {
+ throw new UnsupportedOperationException("GroupAlsoByWindow should not use tagged outputs.");
+ }
+
+ Iterable<WindowedValue<KV<K, Iterable<V>>>> getOutputs() {
+ return outputs;
+ }
+ }
+}
[beam] 04/45: Build fixup: enable UsesFailureMessage and UsesSchema
categories on validatesRunner tests
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit ff50759ece9375ad3fc51c8b07b622df38bc19b5
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Thu May 9 17:23:53 2019 +0200
Build fixup: enable UsesFailureMessage and UsesSchema categories on validatesRunner tests
---
runners/spark/build.gradle | 3 ---
1 file changed, 3 deletions(-)
diff --git a/runners/spark/build.gradle b/runners/spark/build.gradle
index ad43212..46e5c25 100644
--- a/runners/spark/build.gradle
+++ b/runners/spark/build.gradle
@@ -185,7 +185,6 @@ task validatesStructuredStreamingRunnerBatch(type: Test) {
useJUnit {
includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner'
excludeCategories 'org.apache.beam.sdk.testing.UsesCustomWindowMerging'
- excludeCategories 'org.apache.beam.sdk.testing.UsesFailureMessage'
// Unbounded
excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedPCollections'
excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream'
@@ -208,8 +207,6 @@ task validatesStructuredStreamingRunnerBatch(type: Test) {
// Portability
excludeCategories 'org.apache.beam.sdk.testing.UsesImpulse'
excludeCategories 'org.apache.beam.sdk.testing.UsesCrossLanguageTransforms'
- // Schema
- excludeCategories 'org.apache.beam.sdk.testing.UsesSchema'
}
}
[beam] 12/45: re-enable reduceFnRunner timers for output
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 6c9a3a8cdf4c3b2c91fdbe50153ffed8a080d4db
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue May 28 09:31:28 2019 +0200
re-enable reduceFnRunner timers for output
---
.../batch/functions/GroupAlsoByWindowViaOutputBufferFn.java | 8 +++-----
1 file changed, 3 insertions(+), 5 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/GroupAlsoByWindowViaOutputBufferFn.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/GroupAlsoByWindowViaOutputBufferFn.java
index 2fb08f5..cc65716 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/GroupAlsoByWindowViaOutputBufferFn.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/functions/GroupAlsoByWindowViaOutputBufferFn.java
@@ -27,6 +27,7 @@ import org.apache.beam.runners.core.ReduceFnRunner;
import org.apache.beam.runners.core.StateInternals;
import org.apache.beam.runners.core.StateInternalsFactory;
import org.apache.beam.runners.core.SystemReduceFn;
+import org.apache.beam.runners.core.TimerInternals;
import org.apache.beam.runners.core.UnsupportedSideInputReader;
import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
import org.apache.beam.runners.core.construction.TriggerTranslation;
@@ -100,21 +101,18 @@ public class GroupAlsoByWindowViaOutputBufferFn<K, InputT, W extends BoundedWind
// Finish any pending windows by advancing the input watermark to infinity.
timerInternals.advanceInputWatermark(BoundedWindow.TIMESTAMP_MAX_VALUE);
- // not supported yet
-/*
// Finally, advance the processing time to infinity to fire any timers.
timerInternals.advanceProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
fireEligibleTimers(timerInternals, reduceFnRunner);
-*/
reduceFnRunner.persist();
return outputter.getOutputs().iterator();
}
-/* private void fireEligibleTimers(
+ private void fireEligibleTimers(
InMemoryTimerInternals timerInternals,
ReduceFnRunner<K, InputT, Iterable<InputT>, W> reduceFnRunner)
throws Exception {
@@ -136,7 +134,7 @@ public class GroupAlsoByWindowViaOutputBufferFn<K, InputT, W extends BoundedWind
reduceFnRunner.onTimers(timers);
timers.clear();
}
- }*/
+ }
private static class GABWOutputWindowedValue<K, V>
implements OutputWindowedValue<KV<K, Iterable<V>>> {
[beam] 27/45: Fix bug in the window merging logic
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 11981e566b3250633d0b6f398bd8ecbcf750fad3
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue Jul 2 16:45:52 2019 +0200
Fix bug in the window merging logic
---
.../translation/batch/AggregatorCombinerGlobally.java | 13 +++++++------
1 file changed, 7 insertions(+), 6 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
index 884f51e..771e42c 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/AggregatorCombinerGlobally.java
@@ -63,9 +63,9 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
}
@Override public Iterable<WindowedValue<AccumT>> reduce(Iterable<WindowedValue<AccumT>> accumulators,
- Tuple2<Integer, WindowedValue<InputT>> inputTupe) {
+ Tuple2<Integer, WindowedValue<InputT>> inputTuple) {
- WindowedValue<InputT> input = inputTupe._2;
+ WindowedValue<InputT> input = inputTuple._2;
//concatenate accumulators windows and input windows and merge the windows
Collection<W> inputWindows = (Collection<W>)input.getWindows();
Set<W> windows = collectAccumulatorsWindows(accumulators);
@@ -80,11 +80,11 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
// iterate through the input windows and for each, create an accumulator with the merged window
// associated to it and call addInput with the accumulator.
// Maintain a map of the accumulators for use as output
- Map<W, Tuple2<AccumT, Instant>> mapState = new HashMap<>();
+ Map<W, Tuple2<AccumT, Instant>> windowToAccumAndInstant = new HashMap<>();
for (W inputWindow:inputWindows) {
W mergedWindow = windowToMergeResult.get(inputWindow);
mergedWindow = mergedWindow == null ? inputWindow : mergedWindow;
- Tuple2<AccumT, Instant> accumAndInstant = mapState.get(mergedWindow);
+ Tuple2<AccumT, Instant> accumAndInstant = windowToAccumAndInstant.get(mergedWindow);
// if there is no accumulator associated with this window yet, create one
if (accumAndInstant == null) {
AccumT accum = combineFn.addInput(combineFn.createAccumulator(), input.getValue());
@@ -92,7 +92,6 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
timestampCombiner.assign(
mergedWindow, windowingStrategy.getWindowFn().getOutputTime(input.getTimestamp(), mergedWindow));
accumAndInstant = new Tuple2<>(accum, windowTimestamp);
- mapState.put(mergedWindow, accumAndInstant);
} else {
AccumT updatedAccum =
combineFn.addInput(accumAndInstant._1, input.getValue());
@@ -101,10 +100,12 @@ class AggregatorCombinerGlobally<InputT, AccumT, OutputT, W extends BoundedWindo
windowingStrategy.getWindowFn().getOutputTime(input.getTimestamp(), mergedWindow)));
accumAndInstant = new Tuple2<>(updatedAccum, updatedTimestamp);
}
+ windowToAccumAndInstant.put(mergedWindow, accumAndInstant);
}
// output the accumulators map
+ //TODO perf: to avoid memory allocation it is authorized to modify accumulators argument
List<WindowedValue<AccumT>> result = new ArrayList<>();
- for (Map.Entry<W, Tuple2<AccumT, Instant>> entry : mapState.entrySet()) {
+ for (Map.Entry<W, Tuple2<AccumT, Instant>> entry : windowToAccumAndInstant.entrySet()) {
AccumT accumulator = entry.getValue()._1;
Instant windowTimestamp = entry.getValue()._2;
W window = entry.getKey();
[beam] 40/45: Add metrics support in DoFn
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 12eb126682e2dd6b367ed16dcc2e0930d38981dc
Author: Alexey Romanenko <ar...@gmail.com>
AuthorDate: Fri Jul 5 16:47:56 2019 +0200
Add metrics support in DoFn
---
.../SparkStructuredStreamingPipelineOptions.java | 8 +-
.../SparkStructuredStreamingPipelineResult.java | 7 +-
.../SparkStructuredStreamingRunner.java | 60 ++++++++++-
.../aggregators/AggregatorsAccumulator.java | 70 +++++++++++++
.../aggregators/NamedAggregators.java | 112 +++++++++++++++++++++
.../aggregators/NamedAggregatorsAccumulator.java | 63 ++++++++++++
.../package-info.java} | 11 +-
.../AggregatorMetric.java} | 28 ++++--
.../metrics/AggregatorMetricSource.java | 49 +++++++++
.../CompositeSource.java} | 34 +++++--
.../metrics/MetricsAccumulator.java | 71 +++++++++++++
.../MetricsContainerStepMapAccumulator.java | 65 ++++++++++++
.../metrics/SparkBeamMetric.java | 90 +++++++++++++++++
.../metrics/SparkBeamMetricSource.java | 48 +++++++++
.../SparkMetricsContainerStepMap.java} | 27 +++--
.../package-info.java} | 11 +-
.../translation/batch/DoFnFunction.java | 15 ++-
.../translation/batch/DoFnRunnerWithMetrics.java | 97 ++++++++++++++++++
.../translation/batch/ParDoTranslatorBatch.java | 8 ++
19 files changed, 829 insertions(+), 45 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
index b115d9b..27743bd 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
@@ -18,10 +18,16 @@
package org.apache.beam.runners.spark.structuredstreaming;
import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
/**
* Spark runner {@link PipelineOptions} handles Spark execution-related configurations, such as the
* master address, and other user-related knobs.
*/
-public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipelineOptions {}
+public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("Enable/disable sending aggregator values to Spark's metric sinks")
+ @Default.Boolean(true)
+ Boolean getEnableSparkMetricSinks();
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineResult.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineResult.java
index d0198d4..8ccb97a 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineResult.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineResult.java
@@ -19,10 +19,14 @@ package org.apache.beam.runners.spark.structuredstreaming;
import java.io.IOException;
import javax.annotation.Nullable;
+
+import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.metrics.MetricResults;
import org.joda.time.Duration;
+import static org.apache.beam.runners.core.metrics.MetricsContainerStepMap.asAttemptedOnlyMetricResults;
+
/** Represents a Spark pipeline execution result. */
class SparkStructuredStreamingPipelineResult implements PipelineResult {
@@ -50,9 +54,8 @@ class SparkStructuredStreamingPipelineResult implements PipelineResult {
return null;
}
- @Nullable // TODO: remove once method will be implemented
@Override
public MetricResults metrics() {
- return null;
+ return asAttemptedOnlyMetricResults(MetricsAccumulator.getInstance().value());
}
}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunner.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunner.java
index c63efe4..29e4373 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunner.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingRunner.java
@@ -19,15 +19,26 @@ package org.apache.beam.runners.spark.structuredstreaming;
import static org.apache.beam.runners.core.construction.PipelineResources.detectClassPathResourcesToStage;
+import org.apache.beam.runners.core.metrics.MetricsPusher;
+import org.apache.beam.runners.spark.structuredstreaming.aggregators.AggregatorsAccumulator;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.AggregatorMetricSource;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.CompositeSource;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.SparkBeamMetricSource;
import org.apache.beam.runners.spark.structuredstreaming.translation.PipelineTranslator;
import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.PipelineTranslatorBatch;
import org.apache.beam.runners.spark.structuredstreaming.translation.streaming.PipelineTranslatorStreaming;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineRunner;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+import org.apache.beam.sdk.metrics.MetricsOptions;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.PipelineOptionsValidator;
+import org.apache.spark.SparkEnv$;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.metrics.MetricsSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -114,11 +125,29 @@ public final class SparkStructuredStreamingRunner
@Override
public SparkStructuredStreamingPipelineResult run(final Pipeline pipeline) {
+ MetricsEnvironment.setMetricsSupported(true);
+
+ // clear state of Aggregators, Metrics and Watermarks if exists.
+ AggregatorsAccumulator.clear();
+ MetricsAccumulator.clear();
+
TranslationContext translationContext = translatePipeline(pipeline);
// TODO initialise other services: checkpointing, metrics system, listeners, ...
// TODO pass testMode using pipelineOptions
translationContext.startPipeline(true);
- return new SparkStructuredStreamingPipelineResult();
+
+ SparkStructuredStreamingPipelineResult result = new SparkStructuredStreamingPipelineResult();
+
+ if (options.getEnableSparkMetricSinks()) {
+ registerMetricsSource(options.getAppName());
+ }
+
+ MetricsPusher metricsPusher =
+ new MetricsPusher(
+ MetricsAccumulator.getInstance().value(), options.as(MetricsOptions.class), result);
+ metricsPusher.start();
+
+ return result;
}
private TranslationContext translatePipeline(Pipeline pipeline) {
@@ -129,7 +158,36 @@ public final class SparkStructuredStreamingRunner
options.isStreaming()
? new PipelineTranslatorStreaming(options)
: new PipelineTranslatorBatch(options);
+
+ final JavaSparkContext jsc =
+ JavaSparkContext.fromSparkContext(
+ pipelineTranslator.getTranslationContext().getSparkSession().sparkContext());
+ initAccumulators(options, jsc);
+
pipelineTranslator.translate(pipeline);
return pipelineTranslator.getTranslationContext();
}
+
+ private void registerMetricsSource(String appName) {
+ final MetricsSystem metricsSystem = SparkEnv$.MODULE$.get().metricsSystem();
+ final AggregatorMetricSource aggregatorMetricSource =
+ new AggregatorMetricSource(null, AggregatorsAccumulator.getInstance().value());
+ final SparkBeamMetricSource metricsSource = new SparkBeamMetricSource(null);
+ final CompositeSource compositeSource =
+ new CompositeSource(
+ appName + ".Beam",
+ metricsSource.metricRegistry(),
+ aggregatorMetricSource.metricRegistry());
+ // re-register the metrics in case of context re-use
+ metricsSystem.removeSource(compositeSource);
+ metricsSystem.registerSource(compositeSource);
+ }
+
+ /** Init Metrics/Aggregators accumulators. This method is idempotent. */
+ public static void initAccumulators(
+ SparkStructuredStreamingPipelineOptions opts, JavaSparkContext jsc) {
+ // Init metrics accumulators
+ MetricsAccumulator.init(jsc);
+ AggregatorsAccumulator.init(jsc);
+ }
}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/AggregatorsAccumulator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/AggregatorsAccumulator.java
new file mode 100644
index 0000000..2f31830
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/AggregatorsAccumulator.java
@@ -0,0 +1,70 @@
+/*
+ * 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.spark.structuredstreaming.aggregators;
+
+import org.apache.beam.vendor.guava.v20_0.com.google.common.annotations.VisibleForTesting;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.util.AccumulatorV2;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * For resilience, {@link AccumulatorV2 Accumulators} are required to be wrapped in a Singleton.
+ *
+ * @see <a
+ * href="https://spark.apache.org/docs/latest/api/java/org/apache/spark/util/AccumulatorV2.html">accumulatorsV2</a>
+ */
+public class AggregatorsAccumulator {
+ private static final Logger LOG = LoggerFactory.getLogger(AggregatorsAccumulator.class);
+
+ private static final String ACCUMULATOR_NAME = "Beam.Aggregators";
+
+ private static volatile NamedAggregatorsAccumulator instance = null;
+
+ /** Init aggregators accumulator if it has not been initiated. This method is idempotent. */
+ public static void init(JavaSparkContext jsc) {
+ if (instance == null) {
+ synchronized (AggregatorsAccumulator.class) {
+ if (instance == null) {
+ NamedAggregators namedAggregators = new NamedAggregators();
+ NamedAggregatorsAccumulator accumulator =
+ new NamedAggregatorsAccumulator(namedAggregators);
+ jsc.sc().register(accumulator, ACCUMULATOR_NAME);
+
+ instance = accumulator;
+ }
+ }
+ LOG.info("Instantiated aggregators accumulator: " + instance.value());
+ }
+ }
+
+ public static NamedAggregatorsAccumulator getInstance() {
+ if (instance == null) {
+ throw new IllegalStateException("Aggregrators accumulator has not been instantiated");
+ } else {
+ return instance;
+ }
+ }
+
+ @VisibleForTesting
+ public static void clear() {
+ synchronized (AggregatorsAccumulator.class) {
+ instance = null;
+ }
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/NamedAggregators.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/NamedAggregators.java
new file mode 100644
index 0000000..6fe3659
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/NamedAggregators.java
@@ -0,0 +1,112 @@
+/*
+ * 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.spark.structuredstreaming.aggregators;
+
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableMap;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Maps;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * This class wraps a map of named aggregators. Spark expects that all accumulators be declared
+ * before a job is launched. Beam allows aggregators to be used and incremented on the fly. We
+ * create a map of named aggregators and instantiate in the the spark context before the job is
+ * launched. We can then add aggregators on the fly in Spark.
+ */
+public class NamedAggregators implements Serializable {
+ /** Map from aggregator name to current state. */
+ private final Map<String, State<?, ?, ?>> mNamedAggregators = new TreeMap<>();
+
+ /** Constructs a new NamedAggregators instance. */
+ public NamedAggregators() {}
+
+ /**
+ * @param name Name of aggregator to retrieve.
+ * @param typeClass Type class to cast the value to.
+ * @param <T> Type to be returned.
+ * @return the value of the aggregator associated with the specified name, or <code>null</code> if
+ * the specified aggregator could not be found.
+ */
+ public <T> T getValue(String name, Class<T> typeClass) {
+ final State<?, ?, ?> state = mNamedAggregators.get(name);
+ return state != null ? typeClass.cast(state.render()) : null;
+ }
+
+ /** @return a map of all the aggregator names and their <b>rendered </b>values */
+ public Map<String, ?> renderAll() {
+ return ImmutableMap.copyOf(Maps.transformValues(mNamedAggregators, State::render));
+ }
+
+ /**
+ * Merges another NamedAggregators instance with this instance.
+ *
+ * @param other The other instance of named aggregators ot merge.
+ * @return This instance of Named aggregators with associated states updated to reflect the other
+ * instance's aggregators.
+ */
+ public NamedAggregators merge(
+ NamedAggregators other) {
+ for (Map.Entry<String, State<?, ?, ?>> e : other.mNamedAggregators.entrySet()) {
+ String key = e.getKey();
+ State<?, ?, ?> otherValue = e.getValue();
+ mNamedAggregators.merge(key, otherValue, NamedAggregators::merge);
+ }
+ return this;
+ }
+
+ /**
+ * Helper method to merge States whose generic types aren't provably the same, so require some
+ * casting.
+ */
+ @SuppressWarnings("unchecked")
+ private static <InputT, InterT, OutputT> State<InputT, InterT, OutputT> merge(
+ State<?, ?, ?> s1, State<?, ?, ?> s2) {
+ return ((State<InputT, InterT, OutputT>) s1).merge((State<InputT, InterT, OutputT>) s2);
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ for (Map.Entry<String, State<?, ?, ?>> e : mNamedAggregators.entrySet()) {
+ sb.append(e.getKey()).append(": ").append(e.getValue().render()).append(" ");
+ }
+ return sb.toString();
+ }
+
+ /**
+ * @param <InputT> Input data type
+ * @param <InterT> Intermediate data type (useful for averages)
+ * @param <OutputT> Output data type
+ */
+ public interface State<InputT, InterT, OutputT> extends Serializable {
+
+ /** @param element new element to update state */
+ void update(InputT element);
+
+ State<InputT, InterT, OutputT> merge(State<InputT, InterT, OutputT> other);
+
+ InterT current();
+
+ OutputT render();
+
+ Combine.CombineFn<InputT, InterT, OutputT> getCombineFn();
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/NamedAggregatorsAccumulator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/NamedAggregatorsAccumulator.java
new file mode 100644
index 0000000..992e63f
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/NamedAggregatorsAccumulator.java
@@ -0,0 +1,63 @@
+/*
+ * 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.spark.structuredstreaming.aggregators;
+
+import org.apache.spark.util.AccumulatorV2;
+
+/** {@link AccumulatorV2} implementation for {@link NamedAggregators}. */
+public class NamedAggregatorsAccumulator extends AccumulatorV2<NamedAggregators, NamedAggregators> {
+ private static final NamedAggregators empty = new NamedAggregators();
+
+ private NamedAggregators value;
+
+ public NamedAggregatorsAccumulator(NamedAggregators value) {
+ this.value = value;
+ }
+
+ @Override
+ public boolean isZero() {
+ return value.equals(empty);
+ }
+
+ @Override
+ public NamedAggregatorsAccumulator copy() {
+ NamedAggregators newContainer = new NamedAggregators();
+ newContainer.merge(value);
+ return new NamedAggregatorsAccumulator(newContainer);
+ }
+
+ @Override
+ public void reset() {
+ this.value = new NamedAggregators();
+ }
+
+ @Override
+ public void add(NamedAggregators other) {
+ this.value.merge(other);
+ }
+
+ @Override
+ public void merge(AccumulatorV2<NamedAggregators, NamedAggregators> other) {
+ this.value.merge(other.value());
+ }
+
+ @Override
+ public NamedAggregators value() {
+ return this.value;
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/package-info.java
similarity index 65%
copy from runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
copy to runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/package-info.java
index b115d9b..11a87ee 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/aggregators/package-info.java
@@ -15,13 +15,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.spark.structuredstreaming;
-import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptions;
-
-/**
- * Spark runner {@link PipelineOptions} handles Spark execution-related configurations, such as the
- * master address, and other user-related knobs.
- */
-public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipelineOptions {}
+/** Provides internal utilities for implementing Beam aggregators using Spark accumulators. */
+package org.apache.beam.runners.spark.structuredstreaming.aggregators;
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/AggregatorMetric.java
similarity index 52%
copy from runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
copy to runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/AggregatorMetric.java
index b115d9b..55590a6 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/AggregatorMetric.java
@@ -15,13 +15,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.spark.structuredstreaming;
+package org.apache.beam.runners.spark.structuredstreaming.metrics;
-import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptions;
+import com.codahale.metrics.Metric;
+import org.apache.beam.runners.spark.structuredstreaming.aggregators.NamedAggregators;
-/**
- * Spark runner {@link PipelineOptions} handles Spark execution-related configurations, such as the
- * master address, and other user-related knobs.
- */
-public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipelineOptions {}
+/** An adapter between the {@link NamedAggregators} and Codahale's {@link Metric} interface. */
+public class AggregatorMetric implements Metric {
+
+ private final NamedAggregators namedAggregators;
+
+ private AggregatorMetric(final NamedAggregators namedAggregators) {
+ this.namedAggregators = namedAggregators;
+ }
+
+ public static AggregatorMetric of(final NamedAggregators namedAggregators) {
+ return new AggregatorMetric(namedAggregators);
+ }
+
+ NamedAggregators getNamedAggregators() {
+ return namedAggregators;
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/AggregatorMetricSource.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/AggregatorMetricSource.java
new file mode 100644
index 0000000..406dba3
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/AggregatorMetricSource.java
@@ -0,0 +1,49 @@
+/*
+ * 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.spark.structuredstreaming.metrics;
+
+import com.codahale.metrics.MetricRegistry;
+import org.apache.beam.runners.spark.structuredstreaming.aggregators.NamedAggregators;
+import org.apache.spark.metrics.source.Source;
+
+/**
+ * A Spark {@link Source} that is tailored to expose an {@link AggregatorMetric}, wrapping an
+ * underlying {@link NamedAggregators} instance.
+ */
+public class AggregatorMetricSource implements Source {
+ private static final String METRIC_NAME = "Aggregators";
+
+ private final String name;
+
+ private final MetricRegistry metricRegistry = new MetricRegistry();
+
+ public AggregatorMetricSource(final String name, final NamedAggregators aggregators) {
+ this.name = name;
+ metricRegistry.register(METRIC_NAME, AggregatorMetric.of(aggregators));
+ }
+
+ @Override
+ public String sourceName() {
+ return name;
+ }
+
+ @Override
+ public MetricRegistry metricRegistry() {
+ return metricRegistry;
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/CompositeSource.java
similarity index 50%
copy from runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
copy to runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/CompositeSource.java
index b115d9b..8de06f7 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/CompositeSource.java
@@ -15,13 +15,31 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.spark.structuredstreaming;
+package org.apache.beam.runners.spark.structuredstreaming.metrics;
-import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptions;
+import com.codahale.metrics.MetricRegistry;
+import org.apache.spark.metrics.source.Source;
-/**
- * Spark runner {@link PipelineOptions} handles Spark execution-related configurations, such as the
- * master address, and other user-related knobs.
- */
-public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipelineOptions {}
+/** Composite source made up of several {@link MetricRegistry} instances. */
+public class CompositeSource implements Source {
+ private final String name;
+ private final MetricRegistry metricRegistry;
+
+ public CompositeSource(final String name, MetricRegistry... metricRegistries) {
+ this.name = name;
+ this.metricRegistry = new MetricRegistry();
+ for (MetricRegistry metricRegistry : metricRegistries) {
+ this.metricRegistry.registerAll(metricRegistry);
+ }
+ }
+
+ @Override
+ public String sourceName() {
+ return name;
+ }
+
+ @Override
+ public MetricRegistry metricRegistry() {
+ return metricRegistry;
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java
new file mode 100644
index 0000000..da5f28d
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java
@@ -0,0 +1,71 @@
+/*
+ * 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.spark.structuredstreaming.metrics;
+
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.annotations.VisibleForTesting;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.util.AccumulatorV2;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * For resilience, {@link AccumulatorV2 Accumulators} are required to be wrapped in a Singleton.
+ *
+ * @see <a
+ * href="https://spark.apache.org/docs/latest/api/java/org/apache/spark/util/AccumulatorV2.html">accumulatorsV2</a>
+ */
+public class MetricsAccumulator {
+ private static final Logger LOG = LoggerFactory.getLogger(MetricsAccumulator.class);
+
+ private static final String ACCUMULATOR_NAME = "Beam.Metrics";
+
+ private static volatile MetricsContainerStepMapAccumulator instance = null;
+
+ /** Init metrics accumulator if it has not been initiated. This method is idempotent. */
+ public static void init(JavaSparkContext jsc) {
+ if (instance == null) {
+ synchronized (MetricsAccumulator.class) {
+ if (instance == null) {
+ MetricsContainerStepMap metricsContainerStepMap = new MetricsContainerStepMap();
+ MetricsContainerStepMapAccumulator accumulator =
+ new MetricsContainerStepMapAccumulator(metricsContainerStepMap);
+ jsc.sc().register(accumulator, ACCUMULATOR_NAME);
+
+ instance = accumulator;
+ }
+ }
+ LOG.info("Instantiated metrics accumulator: " + instance.value());
+ }
+ }
+
+ public static MetricsContainerStepMapAccumulator getInstance() {
+ if (instance == null) {
+ throw new IllegalStateException("Metrics accumulator has not been instantiated");
+ } else {
+ return instance;
+ }
+ }
+
+ @VisibleForTesting
+ public static void clear() {
+ synchronized (MetricsAccumulator.class) {
+ instance = null;
+ }
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsContainerStepMapAccumulator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsContainerStepMapAccumulator.java
new file mode 100644
index 0000000..c53ff0e
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsContainerStepMapAccumulator.java
@@ -0,0 +1,65 @@
+/*
+ * 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.spark.structuredstreaming.metrics;
+
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.spark.util.AccumulatorV2;
+
+/** {@link AccumulatorV2} implementation for {@link MetricsContainerStepMap}. */
+public class MetricsContainerStepMapAccumulator
+ extends AccumulatorV2<MetricsContainerStepMap, MetricsContainerStepMap> {
+ private static final MetricsContainerStepMap empty = new MetricsContainerStepMap();
+
+ private MetricsContainerStepMap value;
+
+ public MetricsContainerStepMapAccumulator(MetricsContainerStepMap value) {
+ this.value = value;
+ }
+
+ @Override
+ public boolean isZero() {
+ return value.equals(empty);
+ }
+
+ @Override
+ public MetricsContainerStepMapAccumulator copy() {
+ MetricsContainerStepMap newContainer = new MetricsContainerStepMap();
+ newContainer.updateAll(value);
+ return new MetricsContainerStepMapAccumulator(newContainer);
+ }
+
+ @Override
+ public void reset() {
+ this.value = new MetricsContainerStepMap();
+ }
+
+ @Override
+ public void add(MetricsContainerStepMap other) {
+ this.value.updateAll(other);
+ }
+
+ @Override
+ public void merge(AccumulatorV2<MetricsContainerStepMap, MetricsContainerStepMap> other) {
+ this.value.updateAll(other.value());
+ }
+
+ @Override
+ public MetricsContainerStepMap value() {
+ return this.value;
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkBeamMetric.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkBeamMetric.java
new file mode 100644
index 0000000..9d7d129
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkBeamMetric.java
@@ -0,0 +1,90 @@
+/*
+ * 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.spark.structuredstreaming.metrics;
+
+import com.codahale.metrics.Metric;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.beam.sdk.metrics.DistributionResult;
+import org.apache.beam.sdk.metrics.GaugeResult;
+import org.apache.beam.sdk.metrics.MetricKey;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+
+import static java.util.stream.Collectors.toList;
+import static org.apache.beam.runners.core.metrics.MetricsContainerStepMap.asAttemptedOnlyMetricResults;
+
+/**
+ * An adapter between the {@link MetricsContainerStepMap} and Codahale's {@link Metric} interface.
+ */
+class SparkBeamMetric implements Metric {
+ private static final String ILLEGAL_CHARACTERS = "[^A-Za-z0-9-]";
+
+ Map<String, ?> renderAll() {
+ Map<String, Object> metrics = new HashMap<>();
+ MetricResults metricResults =
+ asAttemptedOnlyMetricResults(MetricsAccumulator.getInstance().value());
+ MetricQueryResults metricQueryResults = metricResults.allMetrics();
+ for (MetricResult<Long> metricResult : metricQueryResults.getCounters()) {
+ metrics.put(renderName(metricResult), metricResult.getAttempted());
+ }
+ for (MetricResult<DistributionResult> metricResult : metricQueryResults.getDistributions()) {
+ DistributionResult result = metricResult.getAttempted();
+ metrics.put(renderName(metricResult) + ".count", result.getCount());
+ metrics.put(renderName(metricResult) + ".sum", result.getSum());
+ metrics.put(renderName(metricResult) + ".min", result.getMin());
+ metrics.put(renderName(metricResult) + ".max", result.getMax());
+ metrics.put(renderName(metricResult) + ".mean", result.getMean());
+ }
+ for (MetricResult<GaugeResult> metricResult : metricQueryResults.getGauges()) {
+ metrics.put(renderName(metricResult), metricResult.getAttempted().getValue());
+ }
+ return metrics;
+ }
+
+ @VisibleForTesting
+ String renderName(MetricResult<?> metricResult) {
+ MetricKey key = metricResult.getKey();
+ MetricName name = key.metricName();
+ String step = key.stepName();
+
+ ArrayList<String> pieces = new ArrayList<>();
+
+ if (step != null) {
+ step = step.replaceAll(ILLEGAL_CHARACTERS, "_");
+ if (step.endsWith("_")) {
+ step = step.substring(0, step.length() - 1);
+ }
+ pieces.add(step);
+ }
+
+ pieces.addAll(
+ ImmutableList.of(name.getNamespace(), name.getName()).stream()
+ .map(str -> str.replaceAll(ILLEGAL_CHARACTERS, "_"))
+ .collect(toList()));
+
+ return String.join(".", pieces);
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkBeamMetricSource.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkBeamMetricSource.java
new file mode 100644
index 0000000..665dbc4
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkBeamMetricSource.java
@@ -0,0 +1,48 @@
+/*
+ * 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.spark.structuredstreaming.metrics;
+
+import com.codahale.metrics.MetricRegistry;
+import org.apache.spark.metrics.source.Source;
+
+/**
+ * A Spark {@link Source} that is tailored to expose a {@link SparkBeamMetric}, wrapping an
+ * underlying {@link org.apache.beam.sdk.metrics.MetricResults} instance.
+ */
+public class SparkBeamMetricSource implements Source {
+ private static final String METRIC_NAME = "Metrics";
+
+ private final String name;
+
+ private final MetricRegistry metricRegistry = new MetricRegistry();
+
+ public SparkBeamMetricSource(final String name) {
+ this.name = name;
+ metricRegistry.register(METRIC_NAME, new SparkBeamMetric());
+ }
+
+ @Override
+ public String sourceName() {
+ return name;
+ }
+
+ @Override
+ public MetricRegistry metricRegistry() {
+ return metricRegistry;
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkMetricsContainerStepMap.java
similarity index 56%
copy from runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
copy to runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkMetricsContainerStepMap.java
index b115d9b..fff68f2 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/SparkMetricsContainerStepMap.java
@@ -15,13 +15,28 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.spark.structuredstreaming;
+package org.apache.beam.runners.spark.structuredstreaming.metrics;
-import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
/**
- * Spark runner {@link PipelineOptions} handles Spark execution-related configurations, such as the
- * master address, and other user-related knobs.
+ * Sole purpose of this class is to override {@link #toString()} of {@link MetricsContainerStepMap}
+ * in order to show meaningful metrics in Spark Web Interface.
*/
-public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipelineOptions {}
+class SparkMetricsContainerStepMap extends MetricsContainerStepMap {
+
+ @Override
+ public String toString() {
+ return new SparkBeamMetric().renderAll().toString();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return super.equals(o);
+ }
+
+ @Override
+ public int hashCode() {
+ return super.hashCode();
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/package-info.java
similarity index 65%
copy from runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
copy to runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/package-info.java
index b115d9b..16a1a95 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/SparkStructuredStreamingPipelineOptions.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/package-info.java
@@ -15,13 +15,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.spark.structuredstreaming;
-import org.apache.beam.runners.spark.SparkCommonPipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptions;
-
-/**
- * Spark runner {@link PipelineOptions} handles Spark execution-related configurations, such as the
- * master address, and other user-related knobs.
- */
-public interface SparkStructuredStreamingPipelineOptions extends SparkCommonPipelineOptions {}
+/** Provides internal utilities for implementing Beam metrics using Spark accumulators. */
+package org.apache.beam.runners.spark.structuredstreaming.metrics;
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnFunction.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnFunction.java
index daa802d..bb69b6d 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnFunction.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnFunction.java
@@ -24,6 +24,7 @@ import java.util.Map;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsContainerStepMapAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.SparkSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.SideInputBroadcast;
@@ -54,6 +55,8 @@ import scala.Tuple2;
public class DoFnFunction<InputT, OutputT>
implements MapPartitionsFunction<WindowedValue<InputT>, Tuple2<TupleTag<?>, WindowedValue<?>>> {
+ private final MetricsContainerStepMapAccumulator metricsAccum;
+ private final String stepName;
private final DoFn<InputT, OutputT> doFn;
private transient boolean wasSetupCalled;
private final WindowingStrategy<?, ?> windowingStrategy;
@@ -67,6 +70,8 @@ public class DoFnFunction<InputT, OutputT>
private DoFnSchemaInformation doFnSchemaInformation;
public DoFnFunction(
+ MetricsContainerStepMapAccumulator metricsAccum,
+ String stepName,
DoFn<InputT, OutputT> doFn,
WindowingStrategy<?, ?> windowingStrategy,
Map<PCollectionView<?>, WindowingStrategy<?, ?>> sideInputs,
@@ -77,6 +82,8 @@ public class DoFnFunction<InputT, OutputT>
Map<TupleTag<?>, Coder<?>> outputCoderMap,
SideInputBroadcast broadcastStateData,
DoFnSchemaInformation doFnSchemaInformation) {
+ this.metricsAccum = metricsAccum;
+ this.stepName = stepName;
this.doFn = doFn;
this.windowingStrategy = windowingStrategy;
this.sideInputs = sideInputs;
@@ -113,7 +120,13 @@ public class DoFnFunction<InputT, OutputT>
windowingStrategy,
doFnSchemaInformation);
- return new ProcessContext<>(doFn, doFnRunner, outputManager, Collections.emptyIterator())
+// DoFnRunnerWithMetrics<InputT, OutputT> doFnRunnerWithMetrics =
+ DoFnRunnerWithMetrics<InputT, OutputT> doFnRunnerWithMetrics =
+ new DoFnRunnerWithMetrics<>(stepName, doFnRunner, metricsAccum);
+
+// return new ProcessContext<>(doFn, doFnRunner, outputManager, Collections.emptyIterator())
+ return new ProcessContext<>(
+ doFn, doFnRunnerWithMetrics, outputManager, Collections.emptyIterator())
.processPartition(iter)
.iterator();
}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnRunnerWithMetrics.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnRunnerWithMetrics.java
new file mode 100644
index 0000000..bc804ab
--- /dev/null
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnRunnerWithMetrics.java
@@ -0,0 +1,97 @@
+/*
+ * 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.spark.structuredstreaming.translation.batch;
+
+import org.apache.beam.runners.core.DoFnRunner;
+import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsContainerStepMapAccumulator;
+import org.apache.beam.sdk.metrics.MetricsContainer;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.joda.time.Instant;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/** DoFnRunner decorator which registers {@link MetricsContainerImpl}. */
+class DoFnRunnerWithMetrics<InputT, OutputT> implements DoFnRunner<InputT, OutputT> {
+ private final DoFnRunner<InputT, OutputT> delegate;
+ private final String stepName;
+ private final MetricsContainerStepMapAccumulator metricsAccum;
+
+ DoFnRunnerWithMetrics(
+ String stepName,
+ DoFnRunner<InputT, OutputT> delegate,
+ MetricsContainerStepMapAccumulator metricsAccum) {
+ this.delegate = delegate;
+ this.stepName = stepName;
+ this.metricsAccum = metricsAccum;
+ }
+
+ @Override
+ public DoFn<InputT, OutputT> getFn() {
+ return delegate.getFn();
+ }
+
+ @Override
+ public void startBundle() {
+ try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(metricsContainer())) {
+ delegate.startBundle();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void processElement(final WindowedValue<InputT> elem) {
+ try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(metricsContainer())) {
+ delegate.processElement(elem);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void onTimer(
+ final String timerId,
+ final BoundedWindow window,
+ final Instant timestamp,
+ final TimeDomain timeDomain) {
+ try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(metricsContainer())) {
+ delegate.onTimer(timerId, window, timestamp, timeDomain);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void finishBundle() {
+ try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(metricsContainer())) {
+ delegate.finishBundle();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private MetricsContainer metricsContainer() {
+ return metricsAccum.value().getContainer(stepName);
+ }
+}
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
index 400b025..b3316e6 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
@@ -25,6 +25,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.beam.runners.core.construction.ParDoTranslation;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
+import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsContainerStepMapAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.TransformTranslator;
import org.apache.beam.runners.spark.structuredstreaming.translation.TranslationContext;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.CoderHelpers;
@@ -62,6 +64,7 @@ class ParDoTranslatorBatch<InputT, OutputT>
@Override
public void translateTransform(
PTransform<PCollection<InputT>, PCollectionTuple> transform, TranslationContext context) {
+ String stepName = context.getCurrentTransform().getFullName();
// Check for not supported advanced features
// TODO: add support of Splittable DoFn
@@ -101,9 +104,14 @@ class ParDoTranslatorBatch<InputT, OutputT>
Map<TupleTag<?>, Coder<?>> outputCoderMap = context.getOutputCoders();
Coder<InputT> inputCoder = ((PCollection<InputT>) context.getInput()).getCoder();
+ MetricsContainerStepMapAccumulator metricsAccum = MetricsAccumulator.getInstance();
+// MetricsContainerStepMapAccumulator metricsAccum = null;
+
@SuppressWarnings("unchecked")
DoFnFunction<InputT, OutputT> doFnWrapper =
new DoFnFunction(
+ metricsAccum,
+ stepName,
doFn,
windowingStrategy,
sideInputStrategies,
[beam] 36/45: add comment in combine globally test
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 777be243f53ae7e7b50a50d6f911e6e410f655b6
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 15:04:59 2019 +0200
add comment in combine globally test
---
.../runners/spark/structuredstreaming/translation/batch/CombineTest.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
index cdce691..b3a2915 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
@@ -51,6 +51,7 @@ public class CombineTest implements Serializable {
PCollection<Integer> input =
p.apply(Create.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)).apply(Sum.integersGlobally());
PAssert.that(input).containsInAnyOrder(55);
+ // uses combine per key
p.run();
}
[beam] 05/45: Pass transform based doFnSchemaInformation in ParDo
translation
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit cb1158ba696fdc0e1abb0cd93c8f457397b60006
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Fri May 10 11:44:53 2019 +0200
Pass transform based doFnSchemaInformation in ParDo translation
---
.../structuredstreaming/translation/batch/ParDoTranslatorBatch.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
index b16d7e9..400b025 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java
@@ -77,6 +77,9 @@ class ParDoTranslatorBatch<InputT, OutputT>
signature.stateDeclarations().size() > 0 || signature.timerDeclarations().size() > 0;
checkState(!stateful, "States and timers are not supported for the moment.");
+ DoFnSchemaInformation doFnSchemaInformation =
+ ParDoTranslation.getSchemaInformation(context.getCurrentTransform());
+
// Init main variables
Dataset<WindowedValue<InputT>> inputDataSet = context.getDataset(context.getInput());
Map<TupleTag<?>, PValue> outputs = context.getOutputs();
@@ -110,7 +113,7 @@ class ParDoTranslatorBatch<InputT, OutputT>
inputCoder,
outputCoderMap,
broadcastStateData,
- DoFnSchemaInformation.create());
+ doFnSchemaInformation);
Dataset<Tuple2<TupleTag<?>, WindowedValue<?>>> allOutputs =
inputDataSet.mapPartitions(doFnWrapper, EncoderHelpers.tuple2Encoder());
[beam] 10/45: Add comment about checkpoint mark
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit ede911864373588507545c404d318133aeeed694
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed May 15 14:24:11 2019 +0200
Add comment about checkpoint mark
---
.../translation/streaming/DatasetSourceStreaming.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/DatasetSourceStreaming.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/DatasetSourceStreaming.java
index 234eecf..1cf52ba 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/DatasetSourceStreaming.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/streaming/DatasetSourceStreaming.java
@@ -134,6 +134,7 @@ class DatasetSourceStreaming implements DataSourceV2, MicroBatchReadSupport {
// offsets are ignored see javadoc
for (DatasetPartitionReader partitionReader : partitionReaders) {
try {
+ // TODO: is checkpointMark stored in reliable storage ?
partitionReader.reader.getCheckpointMark().finalizeCheckpoint();
} catch (IOException e) {
throw new RuntimeException(
[beam] 38/45: Add a test that combine per key preserves windowing
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 5bb44d63f86a9ef82344294d488a52731848c37d
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Jul 4 15:26:41 2019 +0200
Add a test that combine per key preserves windowing
---
.../translation/batch/CombineTest.java | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
index b3a2915..f48f79e 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/CombineTest.java
@@ -27,8 +27,13 @@ import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.Sum;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -69,4 +74,20 @@ public class CombineTest implements Serializable {
PAssert.that(input).containsInAnyOrder(KV.of(1, 9), KV.of(2, 12));
p.run();
}
+
+ @Test
+ public void testCombinePerKeyPreservesWindowing(){
+ PCollection<KV<Integer, Integer>> input = p.apply(Create
+ .timestamped(TimestampedValue.of(KV.of(1, 1), new Instant(1)),
+ TimestampedValue.of(KV.of(1, 3), new Instant(2)),
+ TimestampedValue.of(KV.of(1, 5), new Instant(11)),
+ TimestampedValue.of(KV.of(2, 2), new Instant(3)),
+ TimestampedValue.of(KV.of(2, 4), new Instant(11)),
+ TimestampedValue.of(KV.of(2, 6), new Instant(12))))
+ .apply(Window.into(FixedWindows.of(Duration.millis(10)))).apply(Sum.integersPerKey());
+ PAssert.that(input).containsInAnyOrder(KV.of(1, 4), KV.of(1, 5), KV.of(2, 2), KV.of(2, 10));
+ p.run();
+
+ }
+
}
[beam] 14/45: Build fixup: enable UsesBoundedSplittableParDo
category of tests Since the default overwrite is already in place and it is
breaking only in one subcase
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 41de6daf97f531fe1c05eae30a533abf917dbe69
Author: Ismaël Mejía <ie...@gmail.com>
AuthorDate: Tue May 28 12:26:32 2019 +0200
Build fixup: enable UsesBoundedSplittableParDo category of tests
Since the default overwrite is already in place and it is breaking only
in one subcase
---
runners/spark/build.gradle | 1 -
1 file changed, 1 deletion(-)
diff --git a/runners/spark/build.gradle b/runners/spark/build.gradle
index 9619c5e..7cc0fb7 100644
--- a/runners/spark/build.gradle
+++ b/runners/spark/build.gradle
@@ -203,7 +203,6 @@ task validatesStructuredStreamingRunnerBatch(type: Test) {
excludeCategories 'org.apache.beam.sdk.testing.UsesGaugeMetrics'
excludeCategories 'org.apache.beam.sdk.testing.UsesMetricsPusher'
// SDF
- excludeCategories 'org.apache.beam.sdk.testing.UsesBoundedSplittableParDo'
excludeCategories 'org.apache.beam.sdk.testing.UsesSplittableParDoWithWindowedSideInputs'
excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedSplittableParDo'
// Portability
[beam] 15/45: Add a test that GBK preserves windowing
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 92bf2934f323b4e9d7480576669a3dee459d5b0f
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue May 28 15:23:18 2019 +0200
Add a test that GBK preserves windowing
---
.../translation/batch/GroupByKeyTest.java | 42 ++++++++++++++++++++++
1 file changed, 42 insertions(+)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTest.java
index f608920..c95d07d 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/GroupByKeyTest.java
@@ -29,10 +29,17 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -51,6 +58,41 @@ public class GroupByKeyTest implements Serializable {
}
@Test
+ public void testGroupByKeyPreservesWindowing(){
+ PCollection<KV<Integer, Iterable<Integer>>> input = p.apply(Create
+ .timestamped(
+ TimestampedValue.of(KV.of(1, 1), new Instant(1)),
+ TimestampedValue.of(KV.of(1, 3), new Instant(2)),
+ TimestampedValue.of(KV.of(1, 5), new Instant(11)),
+ TimestampedValue.of(KV.of(2, 2), new Instant(3)),
+ TimestampedValue.of(KV.of(2, 4), new Instant(11)),
+ TimestampedValue.of(KV.of(2, 6), new Instant(12))))
+ .apply(Window.into(FixedWindows.of(Duration.millis(10)))).apply(GroupByKey.create())
+ // do manual assertion for windows because Passert do not support multiple kv with same key (because multiple windows)
+ .apply(ParDo.of(new DoFn<KV<Integer, Iterable<Integer>>, KV<Integer, Iterable<Integer>>>() {
+
+ @ProcessElement public void processElement(ProcessContext context) {
+ KV<Integer, Iterable<Integer>> element = context.element();
+ if (element.getKey() == 1) {
+ if (Iterables.size(element.getValue()) == 2) {
+ assertThat(element.getValue(), containsInAnyOrder(1, 3)); // window [0-10)
+ } else {
+ assertThat(element.getValue(), containsInAnyOrder(5)); // window [10-20)
+ }
+ } else { //key == 2
+ if (Iterables.size(element.getValue()) == 2) {
+ assertThat(element.getValue(), containsInAnyOrder(4, 6)); // window [10-20)
+ } else {
+ assertThat(element.getValue(), containsInAnyOrder(2)); // window [0-10)
+ }
+ }
+ context.output(element);
+ }
+ }));
+ p.run();
+
+ }
+ @Test
public void testGroupByKey() {
List<KV<Integer, Integer>> elems = new ArrayList<>();
elems.add(KV.of(1, 1));