beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [01/12] beam git commit: [BEAM-1180] Implement GearpumpPipelineResult
Date Tue, 24 Jan 2017 03:23:13 GMT
Repository: beam
Updated Branches:
  refs/heads/gearpump-runner 4c445dd0b -> 1ed16f11a


[BEAM-1180] Implement GearpumpPipelineResult


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/21554764
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/21554764
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/21554764

Branch: refs/heads/gearpump-runner
Commit: 21554764056c45ea18be1e844b4ca1bfb71e544a
Parents: 4c445dd
Author: manuzhang <owenzhang1990@gmail.com>
Authored: Tue Dec 20 10:39:56 2016 +0800
Committer: manuzhang <owenzhang1990@gmail.com>
Committed: Wed Jan 4 12:59:08 2017 +0800

----------------------------------------------------------------------
 runners/gearpump/pom.xml                        |  1 +
 .../gearpump/GearpumpPipelineResult.java        | 59 ++++++++++++++++++--
 .../beam/runners/gearpump/GearpumpRunner.java   |  4 +-
 .../runners/gearpump/TestGearpumpRunner.java    |  4 ++
 .../translators/GroupByKeyTranslator.java       |  1 -
 5 files changed, 62 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/21554764/runners/gearpump/pom.xml
----------------------------------------------------------------------
diff --git a/runners/gearpump/pom.xml b/runners/gearpump/pom.xml
index bb35ad7..777ad34 100644
--- a/runners/gearpump/pom.xml
+++ b/runners/gearpump/pom.xml
@@ -69,6 +69,7 @@
                   <dependenciesToScan>
                     <dependency>org.apache.beam:beam-sdks-java-core</dependency>
                   </dependenciesToScan>
+                  <argLine>-noverify</argLine>
                   <excludes>
                     <!-- side input is not supported in Gearpump -->
                     <exclude>

http://git-wip-us.apache.org/repos/asf/beam/blob/21554764/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 ed1201d..9c8f7b3 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
@@ -18,6 +18,7 @@
 package org.apache.beam.runners.gearpump;
 
 import java.io.IOException;
+import java.util.List;
 
 import org.apache.beam.sdk.AggregatorRetrievalException;
 import org.apache.beam.sdk.AggregatorValues;
@@ -26,31 +27,62 @@ 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.MasterToAppMaster.AppMasterData;
+import org.apache.gearpump.cluster.client.ClientContext;
 import org.joda.time.Duration;
 
+import scala.collection.JavaConverters;
+import scala.collection.Seq;
 
 /**
  * Result of executing a {@link Pipeline} with Gearpump.
  */
 public class GearpumpPipelineResult implements PipelineResult {
+
+  private final ClientContext client;
+  private final int appId;
+  private final Duration defaultWaitDuration = Duration.standardSeconds(60);
+  private final Duration defaultWaitInterval = Duration.standardSeconds(10);
+
+  public GearpumpPipelineResult(ClientContext client, int appId) {
+    this.client = client;
+    this.appId = appId;
+  }
+
   @Override
   public State getState() {
-    return null;
+    return getGearpumpState();
   }
 
   @Override
   public State cancel() throws IOException {
-    return null;
+    client.shutdown(appId);
+    return State.CANCELLED;
   }
 
   @Override
   public State waitUntilFinish(Duration duration) {
-    return null;
+    long start = System.currentTimeMillis();
+    do {
+      try {
+        Thread.sleep(defaultWaitInterval.getMillis());
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+    } while (State.RUNNING == getGearpumpState()
+        && (System.currentTimeMillis() - start) < duration.getMillis());
+
+    if (State.RUNNING == getGearpumpState()) {
+      return State.DONE;
+    } else {
+      return State.FAILED;
+    }
   }
 
   @Override
   public State waitUntilFinish() {
-    return null;
+    return waitUntilFinish(defaultWaitDuration);
   }
 
   @Override
@@ -66,4 +98,23 @@ public class GearpumpPipelineResult implements PipelineResult {
     return null;
   }
 
+  private State getGearpumpState() {
+    String status = null;
+    List<AppMasterData> apps =
+        JavaConverters.<AppMasterData>seqAsJavaListConverter(
+            (Seq<AppMasterData>) client.listApps().appMasters()).asJava();
+    for (AppMasterData app: apps) {
+      if (app.appId() == appId) {
+        status = app.status();
+      }
+    }
+    if (null == status || status.equals(MasterToAppMaster.AppMasterNonExist())) {
+      return State.UNKNOWN;
+    } else if (status.equals(MasterToAppMaster.AppMasterActive())) {
+      return State.RUNNING;
+    } else {
+      return State.STOPPED;
+    }
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/21554764/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 4083922..9c44da3 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
@@ -107,9 +107,9 @@ public class GearpumpRunner extends PipelineRunner<GearpumpPipelineResult>
{
     TranslationContext translationContext = new TranslationContext(streamApp, options);
     GearpumpPipelineTranslator translator = new GearpumpPipelineTranslator(translationContext);
     translator.translate(pipeline);
-    streamApp.submit();
+    int appId = streamApp.submit();
 
-    return null;
+    return new GearpumpPipelineResult(clientContext, appId);
   }
 
   private ClientContext getClientContext(GearpumpPipelineOptions options, Config config)
{

http://git-wip-us.apache.org/repos/asf/beam/blob/21554764/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/TestGearpumpRunner.java
----------------------------------------------------------------------
diff --git a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/TestGearpumpRunner.java
b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/TestGearpumpRunner.java
index 89d31a6..ee31fb5 100644
--- a/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/TestGearpumpRunner.java
+++ b/runners/gearpump/src/main/java/org/apache/beam/runners/gearpump/TestGearpumpRunner.java
@@ -19,6 +19,7 @@
 package org.apache.beam.runners.gearpump;
 
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsValidator;
 import org.apache.beam.sdk.runners.PipelineRunner;
@@ -52,7 +53,10 @@ public class TestGearpumpRunner extends PipelineRunner<GearpumpPipelineResult>
{
   @Override
   public GearpumpPipelineResult run(Pipeline pipeline) {
     GearpumpPipelineResult result = delegate.run(pipeline);
+    PipelineResult.State state = result.waitUntilFinish();
     cluster.stop();
+    assert(state == PipelineResult.State.DONE);
+
     return result;
   }
 

http://git-wip-us.apache.org/repos/asf/beam/blob/21554764/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 d64f1bf..989957f 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
@@ -134,7 +134,6 @@ public class GroupByKeyTranslator<K, V> implements TransformTranslator<GroupByKe
   private static class ValueToIterable<K, V>
       implements MapFunction<WindowedValue<KV<K, V>>, WindowedValue<KV<K,
Iterable<V>>>> {
 
-
     @Override
     public WindowedValue<KV<K, Iterable<V>>> apply(WindowedValue<KV<K,
V>> wv) {
       Iterable<V> values = Lists.newArrayList(wv.getValue().getValue());


Mime
View raw message