You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2017/02/15 21:18:05 UTC
[1/2] beam git commit: [BEAM-79] Support merging windows in
GearpumpRunner
Repository: beam
Updated Branches:
refs/heads/gearpump-runner 4001aeb19 -> 2d0aed922
[BEAM-79] Support merging windows in GearpumpRunner
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/7af64720
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/7af64720
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/7af64720
Branch: refs/heads/gearpump-runner
Commit: 7af6472082cbc7f3853e87831ed4bdc72978a3a3
Parents: 4001aeb
Author: manuzhang <ow...@gmail.com>
Authored: Tue Feb 7 22:14:18 2017 +0800
Committer: manuzhang <ow...@gmail.com>
Committed: Wed Feb 15 14:59:42 2017 +0800
----------------------------------------------------------------------
runners/gearpump/pom.xml | 5 -
.../gearpump/GearpumpPipelineResult.java | 8 +-
.../beam/runners/gearpump/GearpumpRunner.java | 24 +---
.../translators/GroupByKeyTranslator.java | 133 +++++++++++--------
.../translators/WindowBoundTranslator.java | 53 +-------
.../gearpump/translators/io/GearpumpSource.java | 6 +-
.../translators/utils/DoFnRunnerFactory.java | 1 +
.../translators/utils/TranslatorUtils.java | 20 +++
.../translators/utils/TranslatorUtilsTest.java | 75 +++++++++++
9 files changed, 186 insertions(+), 139 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/pom.xml
----------------------------------------------------------------------
diff --git a/runners/gearpump/pom.xml b/runners/gearpump/pom.xml
index 7c6fa76..6f91c50 100644
--- a/runners/gearpump/pom.xml
+++ b/runners/gearpump/pom.xml
@@ -93,11 +93,6 @@
org.apache.beam.sdk.transforms.ViewTest,
org.apache.beam.sdk.transforms.join.CoGroupByKeyTest
</exclude>
- <!-- merging windows is not supported in Gearpump -->
- <exclude>
- org.apache.beam.sdk.transforms.windowing.WindowingTest,
- org.apache.beam.sdk.util.ReshuffleTest
- </exclude>
</excludes>
<systemPropertyVariables>
<beamTestPipelineOptions>
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpPipelineResult.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpPipelineResult.java b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpPipelineResult.java
index 9e53517..a3740b7 100644
--- a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpPipelineResult.java
+++ b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpPipelineResult.java
@@ -27,7 +27,7 @@ import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.metrics.MetricResults;
import org.apache.beam.sdk.transforms.Aggregator;
-import org.apache.gearpump.cluster.MasterToAppMaster;
+import org.apache.gearpump.cluster.ApplicationStatus;
import org.apache.gearpump.cluster.MasterToAppMaster.AppMasterData;
import org.apache.gearpump.cluster.client.ClientContext;
import org.joda.time.Duration;
@@ -105,7 +105,7 @@ public class GearpumpPipelineResult implements PipelineResult {
}
private State getGearpumpState() {
- String status = null;
+ ApplicationStatus status = null;
List<AppMasterData> apps =
JavaConverters.<AppMasterData>seqAsJavaListConverter(
(Seq<AppMasterData>) client.listApps().appMasters()).asJava();
@@ -114,9 +114,9 @@ public class GearpumpPipelineResult implements PipelineResult {
status = app.status();
}
}
- if (null == status || status.equals(MasterToAppMaster.AppMasterNonExist())) {
+ if (null == status || status instanceof ApplicationStatus.NONEXIST$) {
return State.UNKNOWN;
- } else if (status.equals(MasterToAppMaster.AppMasterActive())) {
+ } else if (status instanceof ApplicationStatus.ACTIVE$) {
return State.RUNNING;
} else {
return State.STOPPED;
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpRunner.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpRunner.java b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpRunner.java
index 01fdb3b..9ca1eb2 100644
--- a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpRunner.java
+++ b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/GearpumpRunner.java
@@ -29,13 +29,8 @@ import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsValidator;
import org.apache.beam.sdk.runners.PipelineRunner;
import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.transforms.windowing.Window;
-import org.apache.beam.sdk.transforms.windowing.WindowFn;
-import org.apache.beam.sdk.util.IdentityWindowFn;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
@@ -74,11 +69,7 @@ public class GearpumpRunner extends PipelineRunner<GearpumpPipelineResult> {
public <OutputT extends POutput, InputT extends PInput> OutputT apply(
PTransform<InputT, OutputT> transform, InputT input) {
- if (Window.Bound.class.equals(transform.getClass())
- && isNullOrIdentityWindowFn(((Window.Bound) transform).getWindowFn())) {
- return (OutputT) super.apply(
- ParDo.of(new IdentityFn()), input);
- } else if (Flatten.FlattenPCollectionList.class.equals(transform.getClass())
+ if (Flatten.FlattenPCollectionList.class.equals(transform.getClass())
&& ((PCollectionList<?>) input).size() == 0) {
return (OutputT) Pipeline.applyTransform(input.getPipeline().begin(), Create.of());
} else if (Create.Values.class.equals(transform.getClass())) {
@@ -108,7 +99,7 @@ public class GearpumpRunner extends PipelineRunner<GearpumpPipelineResult> {
TranslationContext translationContext = new TranslationContext(streamApp, options);
GearpumpPipelineTranslator translator = new GearpumpPipelineTranslator(translationContext);
translator.translate(pipeline);
- int appId = streamApp.submit();
+ int appId = streamApp.submit().appId();
return new GearpumpPipelineResult(clientContext, appId);
}
@@ -140,15 +131,4 @@ public class GearpumpRunner extends PipelineRunner<GearpumpPipelineResult> {
return config.withValue(GEARPUMP_SERIALIZERS, ConfigValueFactory.fromMap(serializers));
}
- private static class IdentityFn<T> extends DoFn<T, T> {
-
- @ProcessElement
- public void process(ProcessContext c) {
- c.output(c.element());
- }
- }
-
- private boolean isNullOrIdentityWindowFn(WindowFn windowFn) {
- return windowFn == null || windowFn.getClass().equals(IdentityWindowFn.class);
- }
}
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/GroupByKeyTranslator.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/GroupByKeyTranslator.java b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/GroupByKeyTranslator.java
index ac8e218..69a1d11 100644
--- a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/GroupByKeyTranslator.java
+++ b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/GroupByKeyTranslator.java
@@ -23,9 +23,8 @@ import com.google.common.collect.Lists;
import java.io.Serializable;
import java.nio.ByteBuffer;
-import java.time.Instant;
+import java.util.ArrayList;
import java.util.Collection;
-import java.util.LinkedList;
import java.util.List;
import org.apache.beam.runners.gearpump.translators.utils.TranslatorUtils;
@@ -34,9 +33,8 @@ import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.transforms.GroupByKey;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
-import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.transforms.windowing.OutputTimeFn;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
import org.apache.beam.sdk.util.CoderUtils;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
@@ -48,15 +46,14 @@ import org.apache.gearpump.streaming.dsl.javaapi.JavaStream;
import org.apache.gearpump.streaming.dsl.javaapi.functions.GroupByFunction;
import org.apache.gearpump.streaming.dsl.window.api.Discarding$;
import org.apache.gearpump.streaming.dsl.window.api.EventTimeTrigger$;
-import org.apache.gearpump.streaming.dsl.window.api.Window;
-import org.apache.gearpump.streaming.dsl.window.api.WindowFn;
-import org.apache.gearpump.streaming.dsl.window.impl.Bucket;
-import scala.collection.JavaConversions;
-
+import org.apache.gearpump.streaming.dsl.window.api.WindowFunction;
+import org.apache.gearpump.streaming.dsl.window.api.Windows;
+import org.apache.gearpump.streaming.dsl.window.impl.Window;
/**
* {@link GroupByKey} is translated to Gearpump groupBy function.
*/
+@SuppressWarnings({"rawtypes", "unchecked"})
public class GroupByKeyTranslator<K, V> implements TransformTranslator<GroupByKey<K, V>> {
@Override
public void translate(GroupByKey<K, V> transform, TranslationContext context) {
@@ -67,69 +64,51 @@ public class GroupByKeyTranslator<K, V> implements TransformTranslator<GroupByKe
int parallelism = context.getPipelineOptions().getParallelism();
OutputTimeFn<? super BoundedWindow> outputTimeFn = (OutputTimeFn<? super BoundedWindow>)
input.getWindowingStrategy().getOutputTimeFn();
+ WindowFn<KV<K, V>, BoundedWindow> windowFn = (WindowFn<KV<K, V>, BoundedWindow>)
+ input.getWindowingStrategy().getWindowFn();
JavaStream<WindowedValue<KV<K, Iterable<V>>>> outputStream = inputStream
- .window(Window.apply(new GearpumpWindowFn(input.getWindowingStrategy().getWindowFn()),
+ .window(Windows.apply(
+ new GearpumpWindowFn(windowFn.isNonMerging()),
EventTimeTrigger$.MODULE$, Discarding$.MODULE$), "assign_window")
.groupBy(new GroupByFn<K, V>(inputKeyCoder), parallelism, "group_by_Key_and_Window")
.map(new ValueToIterable<K, V>(), "map_value_to_iterable")
- .map(new KeyedByTimestamp<K, V>(), "keyed_by_timestamp")
- .reduce(new Merge<K, V>(outputTimeFn), "merge")
+ .map(new KeyedByTimestamp<K, V>((OutputTimeFn<? super BoundedWindow>)
+ input.getWindowingStrategy().getOutputTimeFn()), "keyed_by_timestamp")
+ .reduce(new Merge<>(windowFn, outputTimeFn), "merge")
.map(new Values<K, V>(), "values");
context.setOutputStream(context.getOutput(transform), outputStream);
}
- private static class GearpumpWindowFn<T, W extends BoundedWindow> implements WindowFn,
- Serializable {
+ private static class GearpumpWindowFn<T, W extends BoundedWindow>
+ implements WindowFunction<WindowedValue<T>>, Serializable {
- private org.apache.beam.sdk.transforms.windowing.WindowFn<T, W> windowFn;
+ private final boolean isNonMerging;
- GearpumpWindowFn(org.apache.beam.sdk.transforms.windowing.WindowFn<T, W> windowFn) {
- this.windowFn = windowFn;
+ public GearpumpWindowFn(boolean isNonMerging) {
+ this.isNonMerging = isNonMerging;
}
@Override
- public scala.collection.immutable.List<Bucket> apply(final Instant timestamp) {
+ public Window[] apply(Context<WindowedValue<T>> context) {
try {
- Collection<W> windows = windowFn.assignWindows(windowFn.new AssignContext() {
- @Override
- public T element() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public org.joda.time.Instant timestamp() {
- return TranslatorUtils.java8TimeToJodaTime(timestamp);
- }
-
- @Override
- public W window() {
- throw new UnsupportedOperationException();
- }
- });
-
- List<Bucket> buckets = new LinkedList<>();
- for (BoundedWindow window : windows) {
- buckets.add(getBucket(window));
- }
- return JavaConversions.asScalaBuffer(buckets).toList();
+ return toGearpumpWindows(context.element().getWindows().toArray(new BoundedWindow[0]));
} catch (Exception e) {
throw new RuntimeException(e);
}
}
- private Bucket getBucket(BoundedWindow window) {
- if (window instanceof IntervalWindow) {
- IntervalWindow intervalWindow = (IntervalWindow) window;
- Instant start = TranslatorUtils.jodaTimeToJava8Time(intervalWindow.start());
- Instant end = TranslatorUtils.jodaTimeToJava8Time(intervalWindow.end());
- return new Bucket(start, end);
- } else if (window instanceof GlobalWindow) {
- Instant end = TranslatorUtils.jodaTimeToJava8Time(window.maxTimestamp());
- return new Bucket(Instant.MIN, end);
- } else {
- throw new RuntimeException("unknown window " + window.getClass().getName());
+ @Override
+ public boolean isNonMerging() {
+ return isNonMerging;
+ }
+
+ private Window[] toGearpumpWindows(BoundedWindow[] windows) {
+ Window[] gwins = new Window[windows.length];
+ for (int i = 0; i < windows.length; i++) {
+ gwins[i] = TranslatorUtils.boundedWindowToGearpumpWindow(windows[i]);
}
+ return gwins;
}
}
@@ -166,19 +145,30 @@ public class GroupByKeyTranslator<K, V> implements TransformTranslator<GroupByKe
extends MapFunction<WindowedValue<KV<K, Iterable<V>>>,
KV<org.joda.time.Instant, WindowedValue<KV<K, Iterable<V>>>>> {
+ private final OutputTimeFn<? super BoundedWindow> outputTimeFn;
+
+ public KeyedByTimestamp(OutputTimeFn<? super BoundedWindow> outputTimeFn) {
+ this.outputTimeFn = outputTimeFn;
+ }
+
@Override
public KV<org.joda.time.Instant, WindowedValue<KV<K, Iterable<V>>>> apply(
WindowedValue<KV<K, Iterable<V>>> wv) {
- return KV.of(wv.getTimestamp(), wv);
+ org.joda.time.Instant timestamp = outputTimeFn.assignOutputTime(wv.getTimestamp(),
+ Iterables.getOnlyElement(wv.getWindows()));
+ return KV.of(timestamp, wv);
}
}
private static class Merge<K, V> extends
ReduceFunction<KV<org.joda.time.Instant, WindowedValue<KV<K, Iterable<V>>>>> {
+ private final WindowFn<KV<K, V>, BoundedWindow> windowFn;
private final OutputTimeFn<? super BoundedWindow> outputTimeFn;
- Merge(OutputTimeFn<? super BoundedWindow> outputTimeFn) {
+ Merge(WindowFn<KV<K, V>, BoundedWindow> windowFn,
+ OutputTimeFn<? super BoundedWindow> outputTimeFn) {
+ this.windowFn = windowFn;
this.outputTimeFn = outputTimeFn;
}
@@ -189,13 +179,40 @@ public class GroupByKeyTranslator<K, V> implements TransformTranslator<GroupByKe
org.joda.time.Instant t1 = kv1.getKey();
org.joda.time.Instant t2 = kv2.getKey();
- WindowedValue<KV<K, Iterable<V>>> wv1 = kv1.getValue();
- WindowedValue<KV<K, Iterable<V>>> wv2 = kv2.getValue();
+ final WindowedValue<KV<K, Iterable<V>>> wv1 = kv1.getValue();
+ final WindowedValue<KV<K, Iterable<V>>> wv2 = kv2.getValue();
+
+ final List<BoundedWindow> mergedWindows = new ArrayList<>();
+ if (!windowFn.isNonMerging()) {
+ try {
+ windowFn.mergeWindows(windowFn.new MergeContext() {
+
+ @Override
+ public Collection<BoundedWindow> windows() {
+ ArrayList<BoundedWindow> windows = new ArrayList<>();
+ windows.addAll(wv1.getWindows());
+ windows.addAll(wv2.getWindows());
+ return windows;
+ }
+
+ @Override
+ public void merge(Collection<BoundedWindow> toBeMerged,
+ BoundedWindow mergeResult) throws Exception {
+ mergedWindows.add(mergeResult);
+ }
+ });
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ mergedWindows.addAll(wv1.getWindows());
+ }
- return KV.of(outputTimeFn.combine(t1, t2),
+ org.joda.time.Instant timestamp = outputTimeFn.combine(t1, t2);
+ return KV.of(timestamp,
WindowedValue.of(KV.of(wv1.getValue().getKey(),
- Iterables.concat(wv1.getValue().getValue(), wv2.getValue().getValue())),
- wv1.getTimestamp(), wv1.getWindows(), wv1.getPane()));
+ Iterables.concat(wv1.getValue().getValue(), wv2.getValue().getValue())), timestamp,
+ mergedWindows, wv1.getPane()));
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/WindowBoundTranslator.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/WindowBoundTranslator.java b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/WindowBoundTranslator.java
index 9bf1936..c0de2df 100644
--- a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/WindowBoundTranslator.java
+++ b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/WindowBoundTranslator.java
@@ -21,23 +21,15 @@ package org.apache.beam.runners.gearpump.translators;
import com.google.common.collect.Iterables;
import java.util.Collection;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.OutputTimeFn;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowingStrategy;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.gearpump.Message;
-import org.apache.gearpump.cluster.UserConfig;
+import org.apache.gearpump.streaming.dsl.api.functions.MapFunction;
import org.apache.gearpump.streaming.dsl.javaapi.JavaStream;
-import org.apache.gearpump.streaming.dsl.javaapi.functions.FlatMapFunction;
-import org.apache.gearpump.streaming.javaapi.Task;
-import org.apache.gearpump.streaming.task.TaskContext;
import org.joda.time.Instant;
/**
@@ -52,34 +44,25 @@ public class WindowBoundTranslator<T> implements TransformTranslator<Window.Bou
JavaStream<WindowedValue<T>> inputStream = context.getInputStream(input);
WindowingStrategy<?, ?> outputStrategy =
transform.getOutputStrategyInternal(input.getWindowingStrategy());
- WindowFn<T, BoundedWindow> windowFn =
- (WindowFn<T, BoundedWindow>) outputStrategy.getWindowFn();
- OutputTimeFn<? super BoundedWindow> outputTimeFn = (OutputTimeFn<? super BoundedWindow>)
- outputStrategy.getOutputTimeFn();
+ WindowFn<T, BoundedWindow> windowFn = (WindowFn<T, BoundedWindow>) outputStrategy.getWindowFn();
JavaStream<WindowedValue<T>> outputStream =
inputStream
- .flatMap(new AssignWindows(windowFn, outputTimeFn), "assign_windows")
- .process(AssignTimestampTask.class, 1, UserConfig.empty(), "assign_timestamp");
+ .map(new AssignWindows(windowFn), "assign_windows");
context.setOutputStream(context.getOutput(transform), outputStream);
}
private static class AssignWindows<T> extends
- FlatMapFunction<WindowedValue<T>, WindowedValue<T>> {
+ MapFunction<WindowedValue<T>, WindowedValue<T>> {
private final WindowFn<T, BoundedWindow> windowFn;
- private final OutputTimeFn<? super BoundedWindow> outputTimeFn;
- AssignWindows(
- WindowFn<T, BoundedWindow> windowFn,
- OutputTimeFn<? super BoundedWindow> outputTimeFn) {
+ AssignWindows(WindowFn<T, BoundedWindow> windowFn) {
this.windowFn = windowFn;
- this.outputTimeFn = outputTimeFn;
}
@Override
- public Iterator<WindowedValue<T>> apply(final WindowedValue<T> value) {
- List<WindowedValue<T>> ret = new LinkedList<>();
+ public WindowedValue<T> apply(final WindowedValue<T> value) {
try {
Collection<BoundedWindow> windows = windowFn.assignWindows(windowFn.new AssignContext() {
@Override
@@ -97,32 +80,10 @@ public class WindowBoundTranslator<T> implements TransformTranslator<Window.Bou
return Iterables.getOnlyElement(value.getWindows());
}
});
- for (BoundedWindow window: windows) {
- Instant timestamp = outputTimeFn.assignOutputTime(value.getTimestamp(), window);
- ret.add(WindowedValue.of(
- value.getValue(), timestamp, window, value.getPane()));
- }
+ return WindowedValue.of(value.getValue(), value.getTimestamp(), windows, value.getPane());
} catch (Exception e) {
throw new RuntimeException(e);
}
- return ret.iterator();
- }
- }
-
- /**
- * Assign WindowedValue timestamp to Gearpump message.
- * @param <T> element type of WindowedValue
- */
- public static class AssignTimestampTask<T> extends Task {
-
- public AssignTimestampTask(TaskContext taskContext, UserConfig userConfig) {
- super(taskContext, userConfig);
- }
-
- @Override
- public void onNext(Message message) {
- final WindowedValue<T> value = (WindowedValue<T>) message.msg();
- context.output(Message.apply(value, value.getTimestamp().getMillis()));
}
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/io/GearpumpSource.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/io/GearpumpSource.java b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/io/GearpumpSource.java
index 6e5b2de..3d0d7c8 100644
--- a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/io/GearpumpSource.java
+++ b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/io/GearpumpSource.java
@@ -62,9 +62,8 @@ public abstract class GearpumpSource<T> implements DataSource {
this.reader = createReader(options);
this.available = reader.start();
} catch (Exception e) {
- throw new RuntimeException(e);
- } finally {
close();
+ throw new RuntimeException(e);
}
}
@@ -81,9 +80,8 @@ public abstract class GearpumpSource<T> implements DataSource {
timestamp.getMillis());
}
} catch (Exception e) {
- throw new RuntimeException(e);
- } finally {
close();
+ throw new RuntimeException(e);
}
return message;
}
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/DoFnRunnerFactory.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/DoFnRunnerFactory.java b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/DoFnRunnerFactory.java
index 7e1402f..aaefb88 100644
--- a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/DoFnRunnerFactory.java
+++ b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/DoFnRunnerFactory.java
@@ -38,6 +38,7 @@ import org.apache.beam.sdk.values.TupleTag;
*/
public class DoFnRunnerFactory<InputT, OutputT> implements Serializable {
+ private static final long serialVersionUID = 1083167395296383469L;
private final DoFn<InputT, OutputT> fn;
private final transient PipelineOptions options;
private final SideInputReader sideInputReader;
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtils.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtils.java b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtils.java
index 9b72275..656fc6a 100644
--- a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtils.java
+++ b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtils.java
@@ -20,6 +20,12 @@ package org.apache.beam.runners.gearpump.translators.utils;
import java.time.Instant;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.gearpump.streaming.dsl.window.impl.Window;
+
+
/**
* Utility methods for translators.
*/
@@ -32,4 +38,18 @@ public class TranslatorUtils {
public static org.joda.time.Instant java8TimeToJodaTime(Instant time) {
return new org.joda.time.Instant(time.toEpochMilli());
}
+
+ public static Window boundedWindowToGearpumpWindow(BoundedWindow window) {
+ Instant end = TranslatorUtils.jodaTimeToJava8Time(window.maxTimestamp().plus(1L));
+ if (window instanceof IntervalWindow) {
+ IntervalWindow intervalWindow = (IntervalWindow) window;
+ Instant start = TranslatorUtils.jodaTimeToJava8Time(intervalWindow.start());
+ return new Window(start, end);
+ } else if (window instanceof GlobalWindow) {
+ return new Window(TranslatorUtils.jodaTimeToJava8Time(BoundedWindow.TIMESTAMP_MIN_VALUE),
+ end);
+ } else {
+ throw new RuntimeException("unknown window " + window.getClass().getName());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/beam/blob/7af64720/runners/gearpump/src/test/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtilsTest.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/test/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtilsTest.java b/runners/gearpump/src/test/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtilsTest.java
new file mode 100644
index 0000000..10976e8
--- /dev/null
+++ b/runners/gearpump/src/test/java/org/apache/beam/runners/gearpump/translators/utils/TranslatorUtilsTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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.gearpump.translators.utils;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+import com.google.common.collect.Lists;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.values.KV;
+import org.apache.gearpump.streaming.dsl.window.impl.Window;
+import org.junit.Test;
+
+
+/**
+ * Tests for {@link TranslatorUtils}.
+ */
+public class TranslatorUtilsTest {
+
+ private static final List<KV<org.joda.time.Instant, Instant>> TEST_VALUES = Lists.newArrayList(
+ KV.of(new org.joda.time.Instant(0), Instant.EPOCH),
+ KV.of(new org.joda.time.Instant(42), Instant.ofEpochMilli(42)),
+ KV.of(new org.joda.time.Instant(Long.MIN_VALUE), Instant.ofEpochMilli(Long.MIN_VALUE)),
+ KV.of(new org.joda.time.Instant(Long.MAX_VALUE), Instant.ofEpochMilli(Long.MAX_VALUE)));
+
+ @Test
+ public void testJodaTimeAndJava8TimeConversion() {
+ for (KV<org.joda.time.Instant, Instant> kv: TEST_VALUES) {
+ assertThat(TranslatorUtils.jodaTimeToJava8Time(kv.getKey()),
+ equalTo(kv.getValue()));
+ assertThat(TranslatorUtils.java8TimeToJodaTime(kv.getValue()),
+ equalTo(kv.getKey()));
+ }
+ }
+
+ @Test
+ public void testBoundedWindowToGearpumpWindow() {
+ assertThat(TranslatorUtils.boundedWindowToGearpumpWindow(
+ new IntervalWindow(new org.joda.time.Instant(0),
+ new org.joda.time.Instant(Long.MAX_VALUE))),
+ equalTo(Window.apply(Instant.EPOCH, Instant.ofEpochMilli(Long.MAX_VALUE))));
+ assertThat(TranslatorUtils.boundedWindowToGearpumpWindow(
+ new IntervalWindow(new org.joda.time.Instant(Long.MIN_VALUE),
+ new org.joda.time.Instant(Long.MAX_VALUE))),
+ equalTo(Window.apply(Instant.ofEpochMilli(Long.MIN_VALUE),
+ Instant.ofEpochMilli(Long.MAX_VALUE))));
+ BoundedWindow globalWindow = GlobalWindow.INSTANCE;
+ assertThat(TranslatorUtils.boundedWindowToGearpumpWindow(globalWindow),
+ equalTo(Window.apply(Instant.ofEpochMilli(Long.MIN_VALUE / 1000),
+ Instant.ofEpochMilli(Long.MAX_VALUE / 1000).minus(Duration.ofDays(1)).plusMillis(1))));
+ }
+}
[2/2] beam git commit: This closes #1935: Support merging windows in
GearpumpRunner
Posted by ke...@apache.org.
This closes #1935: Support merging windows in GearpumpRunner
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/2d0aed92
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/2d0aed92
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/2d0aed92
Branch: refs/heads/gearpump-runner
Commit: 2d0aed9221b4e763917b136ce2ecc60d08621a07
Parents: 4001aeb 7af6472
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Feb 15 13:17:30 2017 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Feb 15 13:17:30 2017 -0800
----------------------------------------------------------------------
runners/gearpump/pom.xml | 5 -
.../gearpump/GearpumpPipelineResult.java | 8 +-
.../beam/runners/gearpump/GearpumpRunner.java | 24 +---
.../translators/GroupByKeyTranslator.java | 133 +++++++++++--------
.../translators/WindowBoundTranslator.java | 53 +-------
.../gearpump/translators/io/GearpumpSource.java | 6 +-
.../translators/utils/DoFnRunnerFactory.java | 1 +
.../translators/utils/TranslatorUtils.java | 20 +++
.../translators/utils/TranslatorUtilsTest.java | 75 +++++++++++
9 files changed, 186 insertions(+), 139 deletions(-)
----------------------------------------------------------------------