beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From p..@apache.org
Subject [2/4] beam git commit: [BEAM-2824] support PipelineResult.waitUntilFinish() in jstorm local mode.
Date Mon, 04 Sep 2017 04:59:26 GMT
[BEAM-2824] support PipelineResult.waitUntilFinish() in jstorm local mode.


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

Branch: refs/heads/jstorm-runner
Commit: df75d8074d7067bf6078289f1f2dfa36548fcd5e
Parents: cda4e62
Author: Pei He <pei@apache.org>
Authored: Tue Aug 29 20:10:06 2017 +0800
Committer: Pei He <pei@apache.org>
Committed: Mon Sep 4 12:57:50 2017 +0800

----------------------------------------------------------------------
 .../beam/runners/jstorm/JStormRunnerResult.java | 44 +++++++++++++++++++-
 .../jstorm/translation/CommonInstance.java      |  5 +++
 .../jstorm/translation/DoFnExecutor.java        |  4 +-
 .../jstorm/translation/ExecutorsBolt.java       | 35 +++++++++++++---
 .../translation/GroupByWindowExecutor.java      |  2 +-
 .../jstorm/translation/MetricsReporter.java     |  6 +--
 .../translation/UnboundedSourceSpout.java       | 38 ++++++++++-------
 7 files changed, 104 insertions(+), 30 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/df75d807/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/JStormRunnerResult.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/JStormRunnerResult.java
b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/JStormRunnerResult.java
index 98d967f..782896e 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/JStormRunnerResult.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/JStormRunnerResult.java
@@ -17,15 +17,26 @@
  */
 package org.apache.beam.runners.jstorm;
 
+import static com.alibaba.jstorm.metric.AsmWindow.M10_WINDOW;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import backtype.storm.Config;
 import backtype.storm.LocalCluster;
+import com.alibaba.jstorm.common.metric.AsmGauge;
+import com.alibaba.jstorm.metric.AsmMetricRegistry;
+import com.alibaba.jstorm.metric.JStormMetrics;
 import com.alibaba.jstorm.utils.JStormUtils;
 import java.io.IOException;
+import java.util.Map;
+import org.apache.beam.runners.jstorm.translation.CommonInstance;
 import org.apache.beam.runners.jstorm.translation.JStormMetricResults;
 import org.apache.beam.sdk.PipelineResult;
 import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.BackOff;
+import org.apache.beam.sdk.util.BackOffUtils;
+import org.apache.beam.sdk.util.FluentBackoff;
+import org.apache.beam.sdk.util.Sleeper;
 import org.joda.time.Duration;
 
 /**
@@ -88,12 +99,21 @@ public abstract class JStormRunnerResult implements PipelineResult {
 
     @Override
     public State waitUntilFinish(Duration duration) {
-      JStormUtils.sleepMs(duration.getMillis());
+      Sleeper sleeper = Sleeper.DEFAULT;
+      BackOff backOff = FluentBackoff.DEFAULT.withMaxCumulativeBackoff(duration).backoff();
       try {
-        return cancel();
+        do {
+          if (globalWatermark() >= BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()) {
+            return State.DONE;
+          }
+        } while (BackOffUtils.next(sleeper, backOff));
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        // Ignore InterruptedException
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
+      return State.RUNNING;
     }
 
     @Override
@@ -105,5 +125,25 @@ public abstract class JStormRunnerResult implements PipelineResult {
     public MetricResults metrics() {
       return new JStormMetricResults();
     }
+
+    private long globalWatermark() {
+      AsmMetricRegistry metricRegistry = JStormMetrics.getTaskMetrics();
+      boolean foundWatermark = false;
+      double min = BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis();
+      for (Map.Entry<String, AsmGauge> entry : metricRegistry.getGauges().entrySet())
{
+        if (entry.getKey().endsWith(CommonInstance.BEAM_OUTPUT_WATERMARK_METRICS)) {
+          foundWatermark = true;
+          double outputWatermark = (double) entry.getValue().getValue(M10_WINDOW);
+          if (outputWatermark < min) {
+            min = outputWatermark;
+          }
+        }
+      }
+      if (foundWatermark) {
+        return (long) min;
+      } else {
+        return BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis();
+      }
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/df75d807/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/CommonInstance.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/CommonInstance.java
b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/CommonInstance.java
index b7154cd..1e4f4ae 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/CommonInstance.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/CommonInstance.java
@@ -25,4 +25,9 @@ public class CommonInstance {
   public static final String VALUE = "Value";
 
   public static final String BEAM_WATERMARK_STREAM_ID = "BEAM_WATERMARK";
+
+  public static final String METRIC_KEY_SEPARATOR = "__";
+  public static final String BEAM_SOURCE_WATERMARK_METRICS = "__beam_source_watermark";
+  public static final String BEAM_INPUT_WATERMARK_METRICS = "__beam_input_watermark";
+  public static final String BEAM_OUTPUT_WATERMARK_METRICS = "__beam_output_watermark";
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/df75d807/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/DoFnExecutor.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/DoFnExecutor.java
b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/DoFnExecutor.java
index 72c386a..1ceaf9e 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/DoFnExecutor.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/DoFnExecutor.java
@@ -120,7 +120,6 @@ class DoFnExecutor<InputT, OutputT> implements Executor {
   protected transient StateTag<WatermarkHoldState> watermarkHoldTag;
   protected transient IKvStoreManager kvStoreManager;
   protected transient DefaultStepContext stepContext;
-  protected transient MetricClient metricClient;
 
   public DoFnExecutor(
       String stepName,
@@ -159,7 +158,7 @@ class DoFnExecutor<InputT, OutputT> implements Executor {
             this.sideOutputTags,
             this.stepContext,
             this.windowingStrategy),
-        MetricsReporter.create(metricClient));
+        MetricsReporter.create(executorsBolt.metricClient()));
   }
 
   protected void initService(ExecutorContext context) {
@@ -170,7 +169,6 @@ class DoFnExecutor<InputT, OutputT> implements Executor {
     stepContext = new DefaultStepContext(timerInternals,
         new JStormStateInternals(
             null, kvStoreManager, executorsBolt.timerService(), internalDoFnExecutorId));
-    metricClient = new MetricClient(executorContext.getTopologyContext());
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/beam/blob/df75d807/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/ExecutorsBolt.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/ExecutorsBolt.java
b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/ExecutorsBolt.java
index 449ecb5..3d58a37 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/ExecutorsBolt.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/ExecutorsBolt.java
@@ -28,6 +28,8 @@ import backtype.storm.tuple.Values;
 import com.alibaba.jstorm.cache.IKvStoreManager;
 import com.alibaba.jstorm.cache.KvStoreManagerFactory;
 import com.alibaba.jstorm.cluster.Common;
+import com.alibaba.jstorm.metric.MetricClient;
+import com.alibaba.jstorm.metrics.Gauge;
 import com.alibaba.jstorm.utils.KryoSerializer;
 import com.google.common.base.Function;
 import com.google.common.base.Joiner;
@@ -60,7 +62,9 @@ public class ExecutorsBolt extends AbstractComponent implements IRichBatchBolt
{
 
   protected transient ExecutorContext executorContext;
 
-  protected transient TimerService timerService;
+  private transient TimerService timerService;
+
+  private transient MetricClient metricClient;
 
   // map from input tag to executor inside bolt
   protected final Map<TupleTag, Executor> inputTagToExecutor = Maps.newHashMap();
@@ -80,7 +84,6 @@ public class ExecutorsBolt extends AbstractComponent implements IRichBatchBolt
{
   protected KryoSerializer<WindowedValue> serializer;
 
   public ExecutorsBolt() {
-
   }
 
   public void setStatefulBolt(boolean isStateful) {
@@ -150,12 +153,13 @@ public class ExecutorsBolt extends AbstractComponent implements IRichBatchBolt
{
 
   @Override
   public void prepare(Map stormConf, TopologyContext context, OutputCollector collector)
{
-    LOG.info("Start to prepare for task-{}", context.getThisTaskId());
+    int taskId = context.getThisTaskId();
+    LOG.info("Start to prepare for task-{}", taskId);
     try {
       this.collector = collector;
 
       // init kv store manager
-      String storeName = String.format("task-%d", context.getThisTaskId());
+      String storeName = String.format("task-%d", taskId);
       String stateStorePath = String.format("%s/beam/%s", context.getWorkerIdDir(), storeName);
       IKvStoreManager kvStoreManager = isStatefulBolt
               ? KvStoreManagerFactory.getKvStoreManagerWithMonitor(
@@ -167,6 +171,23 @@ public class ExecutorsBolt extends AbstractComponent implements IRichBatchBolt
{
       // init time service
       timerService = initTimerService();
 
+      // init metrics
+      metricClient = new MetricClient(executorContext.getTopologyContext());
+      metricClient.registerGauge(
+          context.getThisComponentId() + CommonInstance.BEAM_INPUT_WATERMARK_METRICS,
+          new Gauge<Double>() {
+            @Override
+            public Double getValue() {
+              return (double) timerService.currentInputWatermark();
+            }});
+      metricClient.registerGauge(
+          context.getThisComponentId() + CommonInstance.BEAM_OUTPUT_WATERMARK_METRICS,
+          new Gauge<Double>() {
+            @Override
+            public Double getValue() {
+              return (double) timerService.currentOutputWatermark();
+            }});
+
       // init all internal executors
       for (Executor executor : Sets.newHashSet(inputTagToExecutor.values())) {
         executor.init(executorContext);
@@ -175,7 +196,7 @@ public class ExecutorsBolt extends AbstractComponent implements IRichBatchBolt
{
         }
       }
 
-      this.serializer = new KryoSerializer<WindowedValue>(stormConf);
+      this.serializer = new KryoSerializer<>(stormConf);
 
       LOG.info("ExecutorsBolt finished init. LocalExecutors={}", inputTagToExecutor.values());
       LOG.info("inputTagToExecutor={}", inputTagToExecutor);
@@ -296,6 +317,10 @@ public class ExecutorsBolt extends AbstractComponent implements IRichBatchBolt
{
     return timerService;
   }
 
+  public MetricClient metricClient() {
+    return metricClient;
+  }
+
   public void setTimerService(TimerService service) {
     timerService = service;
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/df75d807/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/GroupByWindowExecutor.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/GroupByWindowExecutor.java
b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/GroupByWindowExecutor.java
index cae1bc3..4455e0f 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/GroupByWindowExecutor.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/GroupByWindowExecutor.java
@@ -132,7 +132,7 @@ class GroupByWindowExecutor<K, V>
             this.stepContext,
             this.windowingStrategy);
     return new DoFnRunnerWithMetrics<>(
-        stepName, doFnRunner, MetricsReporter.create(metricClient));
+        stepName, doFnRunner, MetricsReporter.create(executorsBolt.metricClient()));
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/beam/blob/df75d807/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/MetricsReporter.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/MetricsReporter.java
b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/MetricsReporter.java
index cc8c1f8..1e38d1c 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/MetricsReporter.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/MetricsReporter.java
@@ -107,8 +107,8 @@ class MetricsReporter {
 
   private String getMetricNameString(String prefix, MetricResult<?> metricResult) {
     return prefix
-        + METRIC_KEY_SEPARATOR + metricResult.step()
-        + METRIC_KEY_SEPARATOR + metricResult.name().namespace()
-        + METRIC_KEY_SEPARATOR + metricResult.name().name();
+        + CommonInstance.METRIC_KEY_SEPARATOR + metricResult.step()
+        + CommonInstance.METRIC_KEY_SEPARATOR + metricResult.name().namespace()
+        + CommonInstance.METRIC_KEY_SEPARATOR + metricResult.name().name();
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/df75d807/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/UnboundedSourceSpout.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/UnboundedSourceSpout.java
b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/UnboundedSourceSpout.java
index 627a834..73f1f0d 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/UnboundedSourceSpout.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/UnboundedSourceSpout.java
@@ -23,6 +23,8 @@ import backtype.storm.spout.SpoutOutputCollector;
 import backtype.storm.task.TopologyContext;
 import backtype.storm.topology.IRichSpout;
 import backtype.storm.tuple.Values;
+import com.alibaba.jstorm.metric.MetricClient;
+import com.alibaba.jstorm.metrics.Gauge;
 import com.alibaba.jstorm.utils.KryoSerializer;
 import java.io.IOException;
 import java.util.Map;
@@ -115,28 +117,32 @@ public class UnboundedSourceSpout extends AbstractComponent implements
IRichSpou
 
   @Override
   public synchronized void open(Map conf, TopologyContext context, SpoutOutputCollector collector)
{
-    try {
-      this.collector = collector;
-      this.pipelineOptions =
-          this.serializedOptions.getPipelineOptions().as(JStormPipelineOptions.class);
-
-      createSourceReader(null);
+    this.collector = collector;
+    this.pipelineOptions =
+        this.serializedOptions.getPipelineOptions().as(JStormPipelineOptions.class);
+    this.serializer = new KryoSerializer<>(conf);
+    createSourceReader(null);
+    new MetricClient(context).registerGauge(
+        context.getThisComponentId() + CommonInstance.BEAM_SOURCE_WATERMARK_METRICS,
+        new Gauge<Double>() {
+          @Override
+          public Double getValue() {
+            return (double) reader.getWatermark().getMillis();
+          }});
+  }
 
-      this.serializer = new KryoSerializer<>(conf);
+  public synchronized void createSourceReader(UnboundedSource.CheckpointMark checkpointMark)
{
+    try {
+      if (reader != null) {
+        reader.close();
+      }
+      reader = this.source.createReader(this.pipelineOptions, checkpointMark);
+      hasNextRecord = this.reader.start();
     } catch (IOException e) {
       throw new RuntimeException("Unable to create unbounded reader.", e);
     }
   }
 
-  public synchronized void createSourceReader(UnboundedSource.CheckpointMark checkpointMark)
-      throws IOException {
-    if (reader != null) {
-      reader.close();
-    }
-    reader = this.source.createReader(this.pipelineOptions, checkpointMark);
-    hasNextRecord = this.reader.start();
-  }
-
   @Override
   public synchronized void nextTuple() {
     if (!activated.get()) {


Mime
View raw message