You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by pe...@apache.org on 2017/09/08 06:43:53 UTC
[1/4] beam git commit: jstorm-runner: saves MetricResults in local
JStormRunnerResult after cancelling.
Repository: beam
Updated Branches:
refs/heads/jstorm-runner d2b285122 -> ef70031b7
jstorm-runner: saves MetricResults in local JStormRunnerResult after cancelling.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/56ad7a85
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/56ad7a85
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/56ad7a85
Branch: refs/heads/jstorm-runner
Commit: 56ad7a852acf66ebd1061e276317481629b270c4
Parents: d2b2851
Author: Pei He <pe...@apache.org>
Authored: Wed Sep 6 11:05:51 2017 +0800
Committer: Pei He <pe...@apache.org>
Committed: Fri Sep 8 14:42:27 2017 +0800
----------------------------------------------------------------------
.../beam/runners/jstorm/JStormRunnerResult.java | 38 +++++++++++++++++++-
.../beam/runners/jstorm/TestJStormRunner.java | 18 ----------
.../jstorm/translation/JStormMetricResults.java | 25 +++++++++----
3 files changed, 56 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/56ad7a85/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 3962ca2..8848717 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
@@ -23,10 +23,13 @@ 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.common.metric.AsmMetric;
import com.alibaba.jstorm.metric.AsmMetricRegistry;
import com.alibaba.jstorm.metric.JStormMetrics;
+import com.alibaba.jstorm.task.error.TaskReportErrorAndDie;
import com.alibaba.jstorm.utils.JStormUtils;
import java.io.IOException;
+import java.util.Iterator;
import java.util.Map;
import org.apache.beam.runners.jstorm.translation.CommonInstance;
import org.apache.beam.runners.jstorm.translation.JStormMetricResults;
@@ -62,6 +65,7 @@ public abstract class JStormRunnerResult implements PipelineResult {
this.topologyName = checkNotNull(topologyName, "topologyName");
}
+ @Override
public State getState() {
return null;
}
@@ -79,6 +83,7 @@ public abstract class JStormRunnerResult implements PipelineResult {
private final LocalCluster localCluster;
private final long localModeExecuteTimeSecs;
private boolean cancelled;
+ private MetricResults savedMetricResults;
LocalJStormPipelineResult(
String topologyName,
@@ -89,12 +94,27 @@ public abstract class JStormRunnerResult implements PipelineResult {
this.localCluster = checkNotNull(localCluster, "localCluster");
this.localModeExecuteTimeSecs = localModeExecuteTimeSecs;
this.cancelled = false;
+ this.savedMetricResults = null;
+ }
+
+ @Override
+ public State getState() {
+ if (cancelled) {
+ return State.CANCELLED;
+ } else if (globalWatermark() >= BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()) {
+ return State.DONE;
+ } else {
+ return State.RUNNING;
+ }
}
@Override
public State cancel() throws IOException {
+ savedMetricResults = metrics();
localCluster.killTopology(getTopologyName());
localCluster.shutdown();
+ clearPAssertCount();
+ TaskReportErrorAndDie.setExceptionRecord(null);
JStormUtils.sleepMs(1000);
cancelled = true;
return State.CANCELLED;
@@ -129,7 +149,12 @@ public abstract class JStormRunnerResult implements PipelineResult {
@Override
public MetricResults metrics() {
- return new JStormMetricResults();
+ if (savedMetricResults != null) {
+ return savedMetricResults;
+ }
+ AsmMetricRegistry metricRegistry = JStormMetrics.getTaskMetrics();
+ return new JStormMetricResults(
+ metricRegistry.getCounters(), metricRegistry.getGauges(), metricRegistry.getHistograms());
}
private long globalWatermark() {
@@ -151,5 +176,16 @@ public abstract class JStormRunnerResult implements PipelineResult {
return BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis();
}
}
+
+ private void clearPAssertCount() {
+ AsmMetricRegistry taskMetrics = JStormMetrics.getTaskMetrics();
+ Iterator<Map.Entry<String, AsmMetric>> itr = taskMetrics.getMetrics().entrySet().iterator();
+ while (itr.hasNext()) {
+ Map.Entry<String, AsmMetric> metric = itr.next();
+ if (metric.getKey().contains(getTopologyName())) {
+ itr.remove();
+ }
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/56ad7a85/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java
index b637b7c..b28c127 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java
@@ -19,15 +19,11 @@ package org.apache.beam.runners.jstorm;
import static com.google.common.base.Preconditions.checkNotNull;
-import com.alibaba.jstorm.common.metric.AsmMetric;
-import com.alibaba.jstorm.metric.AsmMetricRegistry;
-import com.alibaba.jstorm.metric.JStormMetrics;
import com.alibaba.jstorm.task.error.TaskReportErrorAndDie;
import com.alibaba.jstorm.utils.JStormUtils;
import com.google.common.base.Optional;
import com.google.common.collect.Maps;
import java.io.IOException;
-import java.util.Iterator;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineRunner;
@@ -111,9 +107,7 @@ public class TestJStormRunner extends PipelineRunner<JStormRunnerResult> {
throw new AssertionError("Assertion checks timed out.");
}
} finally {
- clearPAssertCount();
cancel(result);
- TaskReportErrorAndDie.setExceptionRecord(null);
}
}
@@ -188,18 +182,6 @@ public class TestJStormRunner extends PipelineRunner<JStormRunnerResult> {
}
}
- private void clearPAssertCount() {
- String topologyName = options.getJobName();
- AsmMetricRegistry taskMetrics = JStormMetrics.getTaskMetrics();
- Iterator<Map.Entry<String, AsmMetric>> itr = taskMetrics.getMetrics().entrySet().iterator();
- while (itr.hasNext()) {
- Map.Entry<String, AsmMetric> metric = itr.next();
- if (metric.getKey().contains(topologyName)) {
- itr.remove();
- }
- }
- }
-
private void cancel(JStormRunnerResult result) {
try {
result.cancel();
http://git-wip-us.apache.org/repos/asf/beam/blob/56ad7a85/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java
index dbaa28e..986bf0c 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java
@@ -17,11 +17,12 @@
*/
package org.apache.beam.runners.jstorm.translation;
+import static com.google.common.base.Preconditions.checkNotNull;
+
import com.alibaba.jstorm.common.metric.AsmCounter;
import com.alibaba.jstorm.common.metric.AsmGauge;
-import com.alibaba.jstorm.metric.AsmMetricRegistry;
+import com.alibaba.jstorm.common.metric.AsmHistogram;
import com.alibaba.jstorm.metric.AsmWindow;
-import com.alibaba.jstorm.metric.JStormMetrics;
import com.google.auto.value.AutoValue;
import java.util.ArrayList;
import java.util.List;
@@ -42,12 +43,24 @@ import org.joda.time.Instant;
* Implementation of {@link MetricResults} for the JStorm Runner.
*/
public class JStormMetricResults extends MetricResults {
+
+ private final Map<String, AsmCounter> counterMap;
+ private final Map<String, AsmGauge> gaugeMap;
+ private final Map<String, AsmHistogram> histogramMap;
+
+ public JStormMetricResults(
+ Map<String, AsmCounter> counterMap,
+ Map<String, AsmGauge> gaugeMap,
+ Map<String, AsmHistogram> histogramMap) {
+ this.counterMap = checkNotNull(counterMap, "counterMap");
+ this.gaugeMap = checkNotNull(gaugeMap, "gaugeMap");
+ this.histogramMap = checkNotNull(histogramMap, "histogramMap");
+ }
+
@Override
public MetricQueryResults queryMetrics(MetricsFilter filter) {
- AsmMetricRegistry metricRegistry = JStormMetrics.getTaskMetrics();
-
List<MetricResult<Long>> counters = new ArrayList<>();
- for (Map.Entry<String, AsmCounter> entry : metricRegistry.getCounters().entrySet()) {
+ for (Map.Entry<String, AsmCounter> entry : counterMap.entrySet()) {
MetricKey metricKey = MetricsReporter.toMetricKey(entry.getKey());
if (!MetricFiltering.matches(filter, metricKey)) {
continue;
@@ -60,7 +73,7 @@ public class JStormMetricResults extends MetricResults {
}
List<MetricResult<GaugeResult>> gauges = new ArrayList<>();
- for (Map.Entry<String, AsmGauge> entry : metricRegistry.getGauges().entrySet()) {
+ for (Map.Entry<String, AsmGauge> entry : gaugeMap.entrySet()) {
MetricKey metricKey = MetricsReporter.toMetricKey(entry.getKey());
if (!MetricFiltering.matches(filter, metricKey)) {
continue;
[2/4] beam git commit: jstorm-runner: support distribution metrics,
this is required by nexmark.
Posted by pe...@apache.org.
jstorm-runner: support distribution metrics, this is required by nexmark.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/0c388447
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/0c388447
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/0c388447
Branch: refs/heads/jstorm-runner
Commit: 0c388447813a93ab9afc45af591417a82f8c4b1b
Parents: 56ad7a8
Author: Pei He <pe...@apache.org>
Authored: Wed Sep 6 11:05:05 2017 +0800
Committer: Pei He <pe...@apache.org>
Committed: Fri Sep 8 14:42:28 2017 +0800
----------------------------------------------------------------------
.../beam/runners/jstorm/TestJStormRunner.java | 1 +
.../jstorm/translation/JStormMetricResults.java | 38 +++++++++++--
.../jstorm/translation/MetricsReporter.java | 56 ++++++++++++++++++++
3 files changed, 92 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/0c388447/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java
index b28c127..e9f6337 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/TestJStormRunner.java
@@ -61,6 +61,7 @@ public class TestJStormRunner extends PipelineRunner<JStormRunnerResult> {
private TestJStormRunner(JStormPipelineOptions options) {
this.options = options;
Map conf = Maps.newHashMap();
+ conf.put("topology.metric.sample.rate", 1);
// Default state backend is RocksDB, for the users who could not run RocksDB on local testing
// env, following config is used to configure state backend to memory.
// conf.put(ConfigExtension.KV_STORE_TYPE, KvStoreManagerFactory.KvStoreType.memory.toString());
http://git-wip-us.apache.org/repos/asf/beam/blob/0c388447/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java
----------------------------------------------------------------------
diff --git a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java
index 986bf0c..01d4441 100644
--- a/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java
+++ b/runners/jstorm/src/main/java/org/apache/beam/runners/jstorm/translation/JStormMetricResults.java
@@ -22,7 +22,9 @@ import static com.google.common.base.Preconditions.checkNotNull;
import com.alibaba.jstorm.common.metric.AsmCounter;
import com.alibaba.jstorm.common.metric.AsmGauge;
import com.alibaba.jstorm.common.metric.AsmHistogram;
+import com.alibaba.jstorm.common.metric.snapshot.AsmHistogramSnapshot;
import com.alibaba.jstorm.metric.AsmWindow;
+import com.alibaba.jstorm.metrics.Snapshot;
import com.google.auto.value.AutoValue;
import java.util.ArrayList;
import java.util.List;
@@ -87,7 +89,35 @@ public class JStormMetricResults extends MetricResults {
new Instant(0))));
}
- return JStormMetricQueryResults.create(counters, gauges);
+ List<MetricResult<DistributionResult>> distributions = new ArrayList<>();
+ for (Map.Entry<String, AsmHistogram> entry : histogramMap.entrySet()) {
+ MetricKey metricKey = MetricsReporter.toMetricKey(entry.getKey());
+ if (!MetricFiltering.matches(filter, metricKey)) {
+ continue;
+ }
+ AsmHistogram histogram = entry.getValue();
+ histogram.forceFlush();
+
+ Snapshot snapshot =
+ ((AsmHistogramSnapshot) histogram.getSnapshots().get(AsmWindow.M10_WINDOW)).getSnapshot();
+ // TODO: Sum and count might be under estimated, because JStorm histogram only store a fixed
+ // number of values.
+ long sum = 0;
+ for (long v : snapshot.getValues()) {
+ sum += v;
+ }
+ distributions.add(
+ JStormMetricResult.create(
+ metricKey.metricName(),
+ metricKey.stepName(),
+ DistributionResult.create(
+ sum,
+ snapshot.size(),
+ snapshot.getMin(),
+ snapshot.getMax())));
+ }
+
+ return JStormMetricQueryResults.create(counters, gauges, distributions);
}
@AutoValue
@@ -97,8 +127,10 @@ public class JStormMetricResults extends MetricResults {
public static MetricQueryResults create(
Iterable<MetricResult<Long>> counters,
- Iterable<MetricResult<GaugeResult>> gauges) {
- return new AutoValue_JStormMetricResults_JStormMetricQueryResults(counters, gauges, null);
+ Iterable<MetricResult<GaugeResult>> gauges,
+ Iterable<MetricResult<DistributionResult>> distributions) {
+ return new AutoValue_JStormMetricResults_JStormMetricQueryResults(
+ counters, gauges, distributions);
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/0c388447/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 1e38d1c..7867a83 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
@@ -21,12 +21,14 @@ import static com.google.common.base.Preconditions.checkNotNull;
import static org.apache.beam.runners.core.metrics.MetricsContainerStepMap.asAttemptedOnlyMetricResults;
import com.alibaba.jstorm.common.metric.AsmCounter;
+import com.alibaba.jstorm.common.metric.AsmHistogram;
import com.alibaba.jstorm.metric.MetricClient;
import com.alibaba.jstorm.metrics.Gauge;
import com.google.common.collect.Maps;
import java.util.Map;
import org.apache.beam.runners.core.metrics.MetricKey;
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.MetricName;
import org.apache.beam.sdk.metrics.MetricQueryResults;
@@ -45,6 +47,7 @@ class MetricsReporter {
private final MetricsContainerStepMap metricsContainers = new MetricsContainerStepMap();
private final Map<String, Long> reportedCounters = Maps.newHashMap();
+ private final Map<String, DistributionResult> reportedDistributions = Maps.newHashMap();
private final MetricClient metricClient;
public static MetricsReporter create(MetricClient metricClient) {
@@ -77,6 +80,7 @@ class MetricsReporter {
metricResults.queryMetrics(MetricsFilter.builder().build());
updateCounters(metricQueryResults.counters());
updateGauges(metricQueryResults.gauges());
+ updateDistributions(metricQueryResults.distributions());
}
private void updateCounters(Iterable<MetricResult<Long>> counters) {
@@ -105,6 +109,58 @@ class MetricsReporter {
}
}
+ private void updateDistributions(Iterable<MetricResult<DistributionResult>> distributions) {
+ for (final MetricResult<DistributionResult> distributionResult : distributions) {
+ String metricName = getMetricNameString(COUNTER_PREFIX, distributionResult);
+ AsmHistogram histogram = metricClient.registerHistogram(metricName);
+ DistributionResult distribution = distributionResult.attempted();
+ if (distribution.count() == 0) {
+ return;
+ }
+ DistributionResult oldDistribution = reportedDistributions.get(metricName);
+ reportedDistributions.put(metricName, distribution);
+ Long newMin;
+ Long newMax;
+ long restCount;
+ long restSum;
+ if (oldDistribution == null) {
+ newMin = distribution.min();
+ newMax = distribution.min() != distribution.max() ? distribution.max() : null;
+ restCount = distribution.count();
+ restSum = distribution.sum();
+ } else {
+ newMin = distribution.min() < oldDistribution.min() ? distribution.min() : null;
+ newMax =
+ (distribution.max() > oldDistribution.max() && distribution.min() != distribution.max())
+ ? distribution.max() : null;
+ restCount = distribution.count() - oldDistribution.count();
+ restSum = distribution.sum() - oldDistribution.sum();
+ }
+ if (newMin != null) {
+ histogram.update(newMin);
+ restCount--;
+ restSum -= newMin;
+ }
+ if (newMax != null) {
+ histogram.update(newMax);
+ restCount--;
+ restSum -= newMax;
+ }
+ if (restCount == 0) {
+ return;
+ }
+ long restAvg = restSum / restCount;
+ long restMod = restSum % restCount;
+ for (long i = 0; i < restCount; ++i) {
+ if (i == 0) {
+ histogram.update(restAvg + restMod);
+ } else {
+ histogram.update(restAvg);
+ }
+ }
+ }
+ }
+
private String getMetricNameString(String prefix, MetricResult<?> metricResult) {
return prefix
+ CommonInstance.METRIC_KEY_SEPARATOR + metricResult.step()
[3/4] beam git commit: jstorm-runner: support SourceMetrics.
Posted by pe...@apache.org.
jstorm-runner: support SourceMetrics.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/fec423e5
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/fec423e5
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/fec423e5
Branch: refs/heads/jstorm-runner
Commit: fec423e51148d26495fb8e6d17ac204b161f3069
Parents: 0c38844
Author: Pei He <pe...@apache.org>
Authored: Wed Sep 6 12:37:21 2017 +0800
Committer: Pei He <pe...@apache.org>
Committed: Fri Sep 8 14:42:28 2017 +0800
----------------------------------------------------------------------
runners/jstorm/pom.xml | 1 -
.../translation/UnboundedSourceSpout.java | 30 ++++++++++++++------
2 files changed, 22 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/fec423e5/runners/jstorm/pom.xml
----------------------------------------------------------------------
diff --git a/runners/jstorm/pom.xml b/runners/jstorm/pom.xml
index 681adb5..a433fcb 100644
--- a/runners/jstorm/pom.xml
+++ b/runners/jstorm/pom.xml
@@ -90,7 +90,6 @@
<excludedGroups>
org.apache.beam.sdk.testing.UsesSetState,
org.apache.beam.sdk.testing.UsesSplittableParDo,
- org.apache.beam.sdk.testing.UsesAttemptedMetrics,
org.apache.beam.sdk.testing.UsesCommittedMetrics,
org.apache.beam.sdk.testing.UsesTestStream
</excludedGroups>
http://git-wip-us.apache.org/repos/asf/beam/blob/fec423e5/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 73f1f0d..92d2f24 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
@@ -26,12 +26,14 @@ 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.Closeable;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.beam.runners.jstorm.JStormPipelineOptions;
import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
@@ -49,7 +51,7 @@ import org.slf4j.LoggerFactory;
public class UnboundedSourceSpout extends AbstractComponent implements IRichSpout {
private static final Logger LOG = LoggerFactory.getLogger(UnboundedSourceSpout.class);
- private final String name;
+ private final String stepName;
private final String description;
private final UnboundedSource source;
private final SerializedPipelineOptions serializedOptions;
@@ -58,6 +60,7 @@ public class UnboundedSourceSpout extends AbstractComponent implements IRichSpou
private transient JStormPipelineOptions pipelineOptions;
private transient UnboundedSource.UnboundedReader reader;
private transient SpoutOutputCollector collector;
+ private transient MetricsReporter metricsReporter;
private volatile boolean hasNextRecord;
private AtomicBoolean activated = new AtomicBoolean();
@@ -67,12 +70,12 @@ public class UnboundedSourceSpout extends AbstractComponent implements IRichSpou
private long lastWaterMark = BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis();
public UnboundedSourceSpout(
- String name,
+ String stepName,
String description,
UnboundedSource source,
JStormPipelineOptions options,
TupleTag<?> outputTag) {
- this.name = name;
+ this.stepName = checkNotNull(stepName, "stepName");
this.description = checkNotNull(description, "description");
this.source = checkNotNull(source, "source");
this.serializedOptions = new SerializedPipelineOptions(checkNotNull(options, "options"));
@@ -121,18 +124,23 @@ public class UnboundedSourceSpout extends AbstractComponent implements IRichSpou
this.pipelineOptions =
this.serializedOptions.getPipelineOptions().as(JStormPipelineOptions.class);
this.serializer = new KryoSerializer<>(conf);
- createSourceReader(null);
- new MetricClient(context).registerGauge(
+ // init metrics
+ MetricClient metricClient = new MetricClient(context);
+ metricClient.registerGauge(
context.getThisComponentId() + CommonInstance.BEAM_SOURCE_WATERMARK_METRICS,
new Gauge<Double>() {
@Override
public Double getValue() {
return (double) reader.getWatermark().getMillis();
}});
+ metricsReporter = MetricsReporter.create(metricClient);
+
+ createSourceReader(null);
}
public synchronized void createSourceReader(UnboundedSource.CheckpointMark checkpointMark) {
- try {
+ try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(
+ metricsReporter.getMetricsContainer(stepName))) {
if (reader != null) {
reader.close();
}
@@ -148,11 +156,13 @@ public class UnboundedSourceSpout extends AbstractComponent implements IRichSpou
if (!activated.get()) {
return;
}
- try {
+ try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(
+ metricsReporter.getMetricsContainer(stepName))) {
if (!hasNextRecord) {
hasNextRecord = reader.advance();
}
+ boolean emitted = false;
while (hasNextRecord && activated.get()) {
Object value = reader.getCurrent();
Instant timestamp = reader.getCurrentTimestamp();
@@ -169,10 +179,14 @@ public class UnboundedSourceSpout extends AbstractComponent implements IRichSpou
byte[] immutableValue = serializer.serialize(wv);
collector.emit(outputTag.getId(), new Values(immutableValue));
}
+ emitted = true;
// move to next record
hasNextRecord = reader.advance();
}
+ if (emitted) {
+ metricsReporter.updateMetrics();
+ }
Instant waterMark = reader.getWatermark();
if (waterMark != null && lastWaterMark < waterMark.getMillis()) {
@@ -187,7 +201,7 @@ public class UnboundedSourceSpout extends AbstractComponent implements IRichSpou
}
public String getName() {
- return name;
+ return stepName;
}
public TupleTag getOutputTag() {
[4/4] beam git commit: This closes #3812
Posted by pe...@apache.org.
This closes #3812
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/ef70031b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/ef70031b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/ef70031b
Branch: refs/heads/jstorm-runner
Commit: ef70031b7b0e48ae5750725f8ff6a9cd71f37431
Parents: d2b2851 fec423e
Author: Pei He <pe...@apache.org>
Authored: Fri Sep 8 14:43:37 2017 +0800
Committer: Pei He <pe...@apache.org>
Committed: Fri Sep 8 14:43:37 2017 +0800
----------------------------------------------------------------------
runners/jstorm/pom.xml | 1 -
.../beam/runners/jstorm/JStormRunnerResult.java | 38 +++++++++++-
.../beam/runners/jstorm/TestJStormRunner.java | 19 +-----
.../jstorm/translation/JStormMetricResults.java | 63 +++++++++++++++++---
.../jstorm/translation/MetricsReporter.java | 56 +++++++++++++++++
.../translation/UnboundedSourceSpout.java | 30 +++++++---
6 files changed, 170 insertions(+), 37 deletions(-)
----------------------------------------------------------------------