You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by av...@apache.org on 2017/04/26 17:26:44 UTC
[1/2] beam git commit: [BEAM-1958] Standard IO Metrics in Java SDK
Repository: beam
Updated Branches:
refs/heads/master f3cff3695 -> 30dbaf891
[BEAM-1958] Standard IO Metrics in Java SDK
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/41d52be0
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/41d52be0
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/41d52be0
Branch: refs/heads/master
Commit: 41d52be0ec64c83a79d97bfd3c27eb104b546991
Parents: f3cff36
Author: Aviem Zur <av...@gmail.com>
Authored: Thu Apr 13 20:27:33 2017 +0300
Committer: Aviem Zur <av...@gmail.com>
Committed: Wed Apr 26 19:53:29 2017 +0300
----------------------------------------------------------------------
.../streaming/StreamingSourceMetricsTest.java | 12 +-
.../org/apache/beam/sdk/io/CountingSource.java | 6 +-
.../apache/beam/sdk/metrics/CounterCell.java | 19 +--
.../beam/sdk/metrics/DistributionCell.java | 10 +-
.../org/apache/beam/sdk/metrics/GaugeCell.java | 8 +-
.../org/apache/beam/sdk/metrics/Metric.java | 8 +-
.../org/apache/beam/sdk/metrics/MetricCell.java | 8 +-
.../org/apache/beam/sdk/metrics/Metrics.java | 12 ++
.../beam/sdk/metrics/MetricsContainer.java | 8 +-
.../apache/beam/sdk/metrics/SinkMetrics.java | 49 ++++++++
.../apache/beam/sdk/metrics/SourceMetrics.java | 116 +++++++++++++++++++
.../apache/beam/sdk/metrics/MetricsTest.java | 19 ++-
.../sdk/io/gcp/bigquery/StreamingWriteFn.java | 4 +-
.../sdk/io/gcp/pubsub/PubsubUnboundedSink.java | 5 +-
.../io/gcp/pubsub/PubsubUnboundedSource.java | 4 +-
15 files changed, 228 insertions(+), 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
index 80f7f53..5a4b1b5 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/StreamingSourceMetricsTest.java
@@ -29,9 +29,11 @@ import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.Source;
+import org.apache.beam.sdk.metrics.MetricName;
import org.apache.beam.sdk.metrics.MetricNameFilter;
import org.apache.beam.sdk.metrics.MetricQueryResults;
import org.apache.beam.sdk.metrics.MetricsFilter;
+import org.apache.beam.sdk.metrics.SourceMetrics;
import org.joda.time.Duration;
import org.junit.Rule;
import org.junit.Test;
@@ -41,6 +43,7 @@ import org.junit.Test;
* Verify metrics support for {@link Source Sources} in streaming pipelines.
*/
public class StreamingSourceMetricsTest implements Serializable {
+ private static final MetricName ELEMENTS_READ = SourceMetrics.elementsRead().getName();
// Force streaming pipeline using pipeline rule.
@Rule
@@ -65,10 +68,15 @@ public class StreamingSourceMetricsTest implements Serializable {
.metrics()
.queryMetrics(
MetricsFilter.builder()
- .addNameFilter(MetricNameFilter.named("io", "elementsRead"))
+ .addNameFilter(
+ MetricNameFilter.named(ELEMENTS_READ.namespace(), ELEMENTS_READ.name()))
.build());
assertThat(metrics.counters(), hasItem(
- attemptedMetricsResult("io", "elementsRead", "Read(UnboundedCountingSource)", 1000L)));
+ attemptedMetricsResult(
+ ELEMENTS_READ.namespace(),
+ ELEMENTS_READ.name(),
+ "Read(UnboundedCountingSource)",
+ 1000L)));
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
index b66a8b2..81082e5 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/CountingSource.java
@@ -30,7 +30,7 @@ import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.io.UnboundedSource.UnboundedReader;
import org.apache.beam.sdk.metrics.Counter;
-import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.metrics.SourceMetrics;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.values.PCollection;
@@ -190,7 +190,7 @@ public class CountingSource {
private static class BoundedCountingReader extends OffsetBasedSource.OffsetBasedReader<Long> {
private long current;
- private final Counter elementsRead = Metrics.counter("io", "elementsRead");
+ private final Counter elementsRead = SourceMetrics.elementsRead();
public BoundedCountingReader(OffsetBasedSource<Long> source) {
super(source);
@@ -354,7 +354,7 @@ public class CountingSource {
private Instant currentTimestamp;
private Instant firstStarted;
- private final Counter elementsRead = Metrics.counter("io", "elementsRead");
+ private final Counter elementsRead = SourceMetrics.elementsRead();
public UnboundedCountingReader(UnboundedCountingSource source, CounterMark mark) {
this.source = source;
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/CounterCell.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/CounterCell.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/CounterCell.java
index 93700e6..7ab5ebc 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/CounterCell.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/CounterCell.java
@@ -30,7 +30,7 @@ import org.apache.beam.sdk.annotations.Experimental.Kind;
* indirection.
*/
@Experimental(Kind.METRICS)
-public class CounterCell implements MetricCell<Counter, Long>, Counter {
+public class CounterCell implements MetricCell<Long> {
private final DirtyState dirty = new DirtyState();
private final AtomicLong value = new AtomicLong();
@@ -57,28 +57,11 @@ public class CounterCell implements MetricCell<Counter, Long>, Counter {
return value.get();
}
- @Override
- public Counter getInterface() {
- return this;
- }
-
- @Override
public void inc() {
add(1);
}
- @Override
public void inc(long n) {
add(n);
}
-
- @Override
- public void dec() {
- add(-1);
- }
-
- @Override
- public void dec(long n) {
- add(-n);
- }
}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionCell.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionCell.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionCell.java
index 7f684a8..0f3f6a4 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionCell.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DistributionCell.java
@@ -30,11 +30,11 @@ import org.apache.beam.sdk.annotations.Experimental.Kind;
* of indirection.
*/
@Experimental(Kind.METRICS)
-public class DistributionCell implements MetricCell<Distribution, DistributionData>, Distribution {
+public class DistributionCell implements MetricCell<DistributionData> {
private final DirtyState dirty = new DirtyState();
private final AtomicReference<DistributionData> value =
- new AtomicReference<DistributionData>(DistributionData.EMPTY);
+ new AtomicReference<>(DistributionData.EMPTY);
/**
* Package-visibility because all {@link DistributionCell DistributionCells} should be created by
@@ -43,7 +43,6 @@ public class DistributionCell implements MetricCell<Distribution, DistributionDa
DistributionCell() {}
/** Increment the counter by the given amount. */
- @Override
public void update(long n) {
DistributionData original;
do {
@@ -61,10 +60,5 @@ public class DistributionCell implements MetricCell<Distribution, DistributionDa
public DistributionData getCumulative() {
return value.get();
}
-
- @Override
- public Distribution getInterface() {
- return this;
- }
}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeCell.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeCell.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeCell.java
index 35ae822..6f8e880 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeCell.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/GaugeCell.java
@@ -29,12 +29,11 @@ import org.apache.beam.sdk.annotations.Experimental;
* of indirection.
*/
@Experimental(Experimental.Kind.METRICS)
-public class GaugeCell implements MetricCell<Gauge, GaugeData>, Gauge {
+public class GaugeCell implements MetricCell<GaugeData> {
private final DirtyState dirty = new DirtyState();
private final AtomicReference<GaugeData> gaugeValue = new AtomicReference<>(GaugeData.empty());
- @Override
public void set(long value) {
GaugeData original;
do {
@@ -52,9 +51,4 @@ public class GaugeCell implements MetricCell<Gauge, GaugeData>, Gauge {
public GaugeData getCumulative() {
return gaugeValue.get();
}
-
- @Override
- public Gauge getInterface() {
- return this;
- }
}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metric.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metric.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metric.java
index 37a5f65..dcd8a04 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metric.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metric.java
@@ -21,4 +21,10 @@ package org.apache.beam.sdk.metrics;
/**
* Marker interface for all user-facing metrics.
*/
-public interface Metric { }
+public interface Metric {
+
+ /**
+ * The {@link MetricName} given to this metric.
+ */
+ MetricName getName();
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricCell.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricCell.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricCell.java
index 7cf9710..82e30cb 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricCell.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricCell.java
@@ -24,11 +24,10 @@ import org.apache.beam.sdk.annotations.Experimental.Kind;
* A {@link MetricCell} is used for accumulating in-memory changes to a metric. It represents a
* specific metric name in a single context.
*
- * @param <UserT> The type of the user interface for reporting changes to this cell.
* @param <DataT> The type of metric data stored (and extracted) from this cell.
*/
@Experimental(Kind.METRICS)
-public interface MetricCell<UserT extends Metric, DataT> {
+public interface MetricCell<DataT> {
/**
* Return the {@link DirtyState} tracking whether this metric cell contains uncommitted changes.
@@ -39,9 +38,4 @@ public interface MetricCell<UserT extends Metric, DataT> {
* Return the cumulative value of this metric.
*/
DataT getCumulative();
-
- /**
- * Return the user-facing mutator for this cell.
- */
- UserT getInterface();
}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metrics.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metrics.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metrics.java
index 587241a..9286ea9 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metrics.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Metrics.java
@@ -131,6 +131,10 @@ public class Metrics {
@Override public void dec(long n) {
inc(-1 * n);
}
+
+ @Override public MetricName getName() {
+ return name;
+ }
}
/**
@@ -150,6 +154,10 @@ public class Metrics {
container.getDistribution(name).update(value);
}
}
+
+ @Override public MetricName getName() {
+ return name;
+ }
}
/**
@@ -169,5 +177,9 @@ public class Metrics {
container.getGauge(name).set(value);
}
}
+
+ @Override public MetricName getName() {
+ return name;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsContainer.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsContainer.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsContainer.java
index 5812ec6..fbb0da3 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsContainer.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/MetricsContainer.java
@@ -96,7 +96,7 @@ public class MetricsContainer {
return gauges.get(metricName);
}
- private <UpdateT, CellT extends MetricCell<?, UpdateT>>
+ private <UpdateT, CellT extends MetricCell<UpdateT>>
ImmutableList<MetricUpdate<UpdateT>> extractUpdates(
MetricsMap<MetricName, CellT> cells) {
ImmutableList.Builder<MetricUpdate<UpdateT>> updates = ImmutableList.builder();
@@ -120,8 +120,8 @@ public class MetricsContainer {
extractUpdates(gauges));
}
- private void commitUpdates(MetricsMap<MetricName, ? extends MetricCell<?, ?>> cells) {
- for (MetricCell<?, ?> cell : cells.values()) {
+ private void commitUpdates(MetricsMap<MetricName, ? extends MetricCell<?>> cells) {
+ for (MetricCell<?> cell : cells.values()) {
cell.getDirty().afterCommit();
}
}
@@ -135,7 +135,7 @@ public class MetricsContainer {
commitUpdates(distributions);
}
- private <UpdateT, CellT extends MetricCell<?, UpdateT>>
+ private <UpdateT, CellT extends MetricCell<UpdateT>>
ImmutableList<MetricUpdate<UpdateT>> extractCumulatives(
MetricsMap<MetricName, CellT> cells) {
ImmutableList.Builder<MetricUpdate<UpdateT>> updates = ImmutableList.builder();
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SinkMetrics.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SinkMetrics.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SinkMetrics.java
new file mode 100644
index 0000000..f96b6ac
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SinkMetrics.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.sdk.metrics;
+
+/**
+ * Standard Sink Metrics.
+ */
+public class SinkMetrics {
+
+ private static final String SINK_NAMESPACE = "sink";
+
+ private static final String ELEMENTS_WRITTEN = "elements_written";
+ private static final String BYTES_WRITTEN = "bytes_written";
+
+ private static final Counter ELEMENTS_WRITTEN_COUNTER =
+ Metrics.counter(SINK_NAMESPACE, ELEMENTS_WRITTEN);
+ private static final Counter BYTES_WRITTEN_COUNTER =
+ Metrics.counter(SINK_NAMESPACE, BYTES_WRITTEN);
+
+ /**
+ * Counter of elements written to a sink.
+ */
+ public static Counter elementsWritten() {
+ return ELEMENTS_WRITTEN_COUNTER;
+ }
+
+ /**
+ * Counter of bytes written to a sink.
+ */
+ public static Counter bytesWritten() {
+ return BYTES_WRITTEN_COUNTER;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java
new file mode 100644
index 0000000..4479f3a
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java
@@ -0,0 +1,116 @@
+/*
+ * 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.sdk.metrics;
+
+import com.google.common.base.Joiner;
+
+/**
+ * Standard {@link org.apache.beam.sdk.io.Source} Metrics.
+ */
+public class SourceMetrics {
+
+ private static final String SOURCE_NAMESPACE = "source";
+ private static final String SOURCE_SPLITS_NAMESPACE = "source.splits";
+ private static final String SEPARATOR = ".";
+
+ private static final String ELEMENTS_READ = "elements_read";
+ private static final String BYTES_READ = "bytes_read";
+ private static final String BACKLOG_BYTES = "backlog_bytes";
+ private static final String BACKLOG_ELEMENTS = "backlog_elements";
+
+ private static final Counter ELEMENTS_READ_COUNTER =
+ Metrics.counter(SOURCE_NAMESPACE, ELEMENTS_READ);
+ private static final Counter BYTES_READ_COUNTER =
+ Metrics.counter(SOURCE_NAMESPACE, BYTES_READ);
+ private static final Gauge BACKLOG_BYTES_GAUGE =
+ Metrics.gauge(SOURCE_NAMESPACE, BACKLOG_BYTES);
+ private static final Gauge BACKLOG_ELEMENTS_GAUGE =
+ Metrics.gauge(SOURCE_NAMESPACE, BACKLOG_ELEMENTS);
+
+ /**
+ * Counter of elements read by a source.
+ */
+ public static Counter elementsRead() {
+ return ELEMENTS_READ_COUNTER;
+ }
+
+ /**
+ * Counter of elements read by a source split.
+ *
+ * <p>Should only be used when there is a small, fixed set of split IDs so as not to overload
+ * metrics backends.</p>
+ */
+ public static Counter elementsReadBySplit(String splitId) {
+ return Metrics.counter(SOURCE_SPLITS_NAMESPACE, renderName(splitId, ELEMENTS_READ));
+ }
+
+ /**
+ * Counter of bytes read by a source.
+ */
+ public static Counter bytesRead() {
+ return BYTES_READ_COUNTER;
+ }
+
+ /**
+ * Counter of bytes read by a source split.
+ *
+ * <p>Should only be used when there is a small, fixed set of split IDs so as not to overload
+ * metrics backends.</p>
+ */
+ public static Counter bytesReadBySplit(String splitId) {
+ return Metrics.counter(SOURCE_SPLITS_NAMESPACE, renderName(splitId, BYTES_READ));
+ }
+
+ /**
+ * Gauge for source backlog in bytes.
+ */
+ public static Gauge backlogBytes() {
+ return BACKLOG_BYTES_GAUGE;
+ }
+
+ /**
+ * Gauge for source split backlog in bytes.
+ *
+ * <p>Should only be used when there is a small, fixed set of split IDs so as not to overload
+ * metrics backends.</p>
+ */
+ public static Gauge backlogBytesOfSplit(String splitId) {
+ return Metrics.gauge(SOURCE_SPLITS_NAMESPACE, renderName(splitId, BACKLOG_BYTES));
+ }
+
+ /**
+ * Gauge for source backlog in elements.
+ */
+ public static Gauge backlogElements() {
+ return BACKLOG_ELEMENTS_GAUGE;
+ }
+
+ /**
+ * Gauge for source split backlog in elements.
+ *
+ * <p>Should only be used when there is a small, fixed set of split IDs so as not to overload
+ * metrics backends.</p>
+ */
+ public static Gauge backlogElementsOfSplit(String splitId) {
+ return Metrics.gauge(SOURCE_SPLITS_NAMESPACE, renderName(splitId, BACKLOG_ELEMENTS));
+ }
+
+ private static String renderName(String... nameParts) {
+ return Joiner.on(SEPARATOR).join(nameParts);
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
index a093e89..c7068e1 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
@@ -56,6 +56,7 @@ public class MetricsTest implements Serializable {
private static final String NAME = "name";
private static final MetricName METRIC_NAME = MetricName.named(NS, NAME);
private static final String NAMESPACE = MetricsTest.class.getName();
+ private static final MetricName ELEMENTS_READ = SourceMetrics.elementsRead().getName();
@Rule
public final transient TestPipeline pipeline = TestPipeline.create();
@@ -245,11 +246,16 @@ public class MetricsTest implements Serializable {
.metrics()
.queryMetrics(
MetricsFilter.builder()
- .addNameFilter(MetricNameFilter.named("io", "elementsRead"))
+ .addNameFilter(
+ MetricNameFilter.named(ELEMENTS_READ.namespace(), ELEMENTS_READ.name()))
.build());
assertThat(metrics.counters(), hasItem(
- attemptedMetricsResult("io", "elementsRead", "Read(BoundedCountingSource)", 1000L)));
+ attemptedMetricsResult(
+ ELEMENTS_READ.namespace(),
+ ELEMENTS_READ.name(),
+ "Read(BoundedCountingSource)",
+ 1000L)));
}
@Test
@@ -269,10 +275,15 @@ public class MetricsTest implements Serializable {
.metrics()
.queryMetrics(
MetricsFilter.builder()
- .addNameFilter(MetricNameFilter.named("io", "elementsRead"))
+ .addNameFilter(
+ MetricNameFilter.named(ELEMENTS_READ.namespace(), ELEMENTS_READ.name()))
.build());
assertThat(metrics.counters(), hasItem(
- attemptedMetricsResult("io", "elementsRead", "Read(UnboundedCountingSource)", 1000L)));
+ attemptedMetricsResult(
+ ELEMENTS_READ.namespace(),
+ ELEMENTS_READ.name(),
+ "Read(UnboundedCountingSource)",
+ 1000L)));
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteFn.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteFn.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteFn.java
index 2b4cd71..fd5f396 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteFn.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteFn.java
@@ -26,7 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.beam.sdk.metrics.Counter;
-import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.metrics.SinkMetrics;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.util.SystemDoFnInternal;
@@ -48,7 +48,7 @@ class StreamingWriteFn
private transient Map<String, List<String>> uniqueIdsForTableRows;
/** Tracks bytes written, exposed as "ByteCount" Counter. */
- private Counter byteCounter = Metrics.counter(StreamingWriteFn.class, "ByteCount");
+ private Counter byteCounter = SinkMetrics.bytesWritten();
StreamingWriteFn(BigQueryServices bqServices) {
this.bqServices = bqServices;
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java
index cf43ae6..002e979 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSink.java
@@ -47,6 +47,7 @@ import org.apache.beam.sdk.io.gcp.pubsub.PubsubClient.PubsubClientFactory;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubClient.TopicPath;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.metrics.SinkMetrics;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.GroupByKey;
@@ -235,8 +236,8 @@ public class PubsubUnboundedSink<T> extends PTransform<PCollection<T>, PDone> {
private transient PubsubClient pubsubClient;
private final Counter batchCounter = Metrics.counter(WriterFn.class, "batches");
- private final Counter elementCounter = Metrics.counter(WriterFn.class, "elements");
- private final Counter byteCounter = Metrics.counter(WriterFn.class, "bytes");
+ private final Counter elementCounter = SinkMetrics.elementsWritten();
+ private final Counter byteCounter = SinkMetrics.bytesWritten();
WriterFn(
PubsubClientFactory pubsubFactory, ValueProvider<TopicPath> topic,
http://git-wip-us.apache.org/repos/asf/beam/blob/41d52be0/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java
index 4979939..b16b665 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java
@@ -59,7 +59,7 @@ import org.apache.beam.sdk.io.gcp.pubsub.PubsubClient.PubsubClientFactory;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubClient.SubscriptionPath;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubClient.TopicPath;
import org.apache.beam.sdk.metrics.Counter;
-import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.metrics.SourceMetrics;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.transforms.Combine;
@@ -1197,7 +1197,7 @@ public class PubsubUnboundedSource<T> extends PTransform<PBegin, PCollection<T>>
// ================================================================================
private static class StatsFn<T> extends DoFn<T, T> {
- private final Counter elementCounter = Metrics.counter(StatsFn.class, "elements");
+ private final Counter elementCounter = SourceMetrics.elementsRead();
private final PubsubClientFactory pubsubFactory;
@Nullable
[2/2] beam git commit: This closes #2538
Posted by av...@apache.org.
This closes #2538
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/30dbaf89
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/30dbaf89
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/30dbaf89
Branch: refs/heads/master
Commit: 30dbaf8913eda2b91c8fc43823f7e5feee17528a
Parents: f3cff36 41d52be
Author: Aviem Zur <av...@gmail.com>
Authored: Wed Apr 26 20:26:24 2017 +0300
Committer: Aviem Zur <av...@gmail.com>
Committed: Wed Apr 26 20:26:24 2017 +0300
----------------------------------------------------------------------
.../streaming/StreamingSourceMetricsTest.java | 12 +-
.../org/apache/beam/sdk/io/CountingSource.java | 6 +-
.../apache/beam/sdk/metrics/CounterCell.java | 19 +--
.../beam/sdk/metrics/DistributionCell.java | 10 +-
.../org/apache/beam/sdk/metrics/GaugeCell.java | 8 +-
.../org/apache/beam/sdk/metrics/Metric.java | 8 +-
.../org/apache/beam/sdk/metrics/MetricCell.java | 8 +-
.../org/apache/beam/sdk/metrics/Metrics.java | 12 ++
.../beam/sdk/metrics/MetricsContainer.java | 8 +-
.../apache/beam/sdk/metrics/SinkMetrics.java | 49 ++++++++
.../apache/beam/sdk/metrics/SourceMetrics.java | 116 +++++++++++++++++++
.../apache/beam/sdk/metrics/MetricsTest.java | 19 ++-
.../sdk/io/gcp/bigquery/StreamingWriteFn.java | 4 +-
.../sdk/io/gcp/pubsub/PubsubUnboundedSink.java | 5 +-
.../io/gcp/pubsub/PubsubUnboundedSource.java | 4 +-
15 files changed, 228 insertions(+), 60 deletions(-)
----------------------------------------------------------------------