You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2025/01/29 17:26:17 UTC
(beam) branch post-2-62-0 created (now 6ee5a50ff06)
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a change to branch post-2-62-0
in repository https://gitbox.apache.org/repos/asf/beam.git
at 6ee5a50ff06 Revert "Add ability to encode and decode histogram data to portable runners (#33013)"
This branch includes the following new commits:
new 6ee5a50ff06 Revert "Add ability to encode and decode histogram data to portable runners (#33013)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(beam) 01/01: Revert "Add ability to encode and decode histogram data to portable runners (#33013)"
Posted by yh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch post-2-62-0
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 6ee5a50ff06cefa7ee31c088632fdf929e328968
Author: Kenneth Knowles <kl...@google.com>
AuthorDate: Fri Jan 24 13:55:26 2025 -0500
Revert "Add ability to encode and decode histogram data to portable runners (#33013)"
This reverts commit 5bfacc935f8510bdf287d8c87b810e08ab9a79ae. The change caused the
core SDK, the SDK harness, and the shared runners-core library to depend on Dataflow.
---
runners/core-java/build.gradle | 2 -
.../core/metrics/MonitoringInfoEncodings.java | 111 +--------------------
.../core/metrics/MonitoringInfoEncodingsTest.java | 47 +--------
sdks/java/core/build.gradle | 1 -
.../org/apache/beam/sdk/util/HistogramData.java | 78 ---------------
sdks/java/harness/build.gradle | 2 -
6 files changed, 2 insertions(+), 239 deletions(-)
diff --git a/runners/core-java/build.gradle b/runners/core-java/build.gradle
index 0f55c10b97f..b477dde9121 100644
--- a/runners/core-java/build.gradle
+++ b/runners/core-java/build.gradle
@@ -42,7 +42,6 @@ dependencies {
implementation project(path: ":model:pipeline", configuration: "shadow")
implementation project(path: ":sdks:java:core", configuration: "shadow")
implementation project(path: ":model:job-management", configuration: "shadow")
- implementation library.java.google_api_services_dataflow
implementation library.java.vendored_guava_32_1_2_jre
implementation library.java.joda_time
implementation library.java.vendored_grpc_1_60_1
@@ -53,6 +52,5 @@ dependencies {
testImplementation library.java.junit
testImplementation library.java.mockito_core
testImplementation library.java.slf4j_api
- testImplementation(library.java.google_api_services_dataflow)
testRuntimeOnly library.java.slf4j_simple
}
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodings.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodings.java
index 88136a864d0..44995b979dd 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodings.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodings.java
@@ -17,19 +17,8 @@
*/
package org.apache.beam.runners.core.metrics;
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.api.services.dataflow.model.Base2Exponent;
-import com.google.api.services.dataflow.model.BucketOptions;
-import com.google.api.services.dataflow.model.DataflowHistogramValue;
-import com.google.api.services.dataflow.model.Linear;
import java.io.IOException;
import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
import java.util.Set;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.DoubleCoder;
@@ -37,14 +26,10 @@ import org.apache.beam.sdk.coders.IterableCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.util.ByteStringOutputStream;
-import org.apache.beam.sdk.util.HistogramData;
-import org.apache.beam.vendor.grpc.v1p60p1.com.google.protobuf.ByteString;
+import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.ByteString;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets;
import org.joda.time.Instant;
-// TODO(#33093): Refactor out DataflowHistogramValue to be runner agnostic, and rename to
-// remove Dataflow reference.
-
/** A set of functions used to encode and decode common monitoring info types. */
public class MonitoringInfoEncodings {
private static final Coder<Long> VARINT_CODER = VarLongCoder.of();
@@ -178,98 +163,4 @@ public class MonitoringInfoEncodings {
throw new RuntimeException(e);
}
}
-
- /** Encodes to {@link MonitoringInfoConstants.TypeUrns#PER_WORKER_HISTOGRAM}. */
- public static ByteString encodeInt64Histogram(HistogramData inputHistogram) {
- try {
- int numberOfBuckets = inputHistogram.getBucketType().getNumBuckets();
-
- DataflowHistogramValue outputHistogram2 = new DataflowHistogramValue();
-
- if (inputHistogram.getBucketType() instanceof HistogramData.LinearBuckets) {
- HistogramData.LinearBuckets buckets =
- (HistogramData.LinearBuckets) inputHistogram.getBucketType();
- Linear linear = new Linear();
- linear.setNumberOfBuckets(numberOfBuckets);
- linear.setWidth(buckets.getWidth());
- linear.setStart(buckets.getStart());
- outputHistogram2.setBucketOptions(new BucketOptions().setLinear(linear));
- } else if (inputHistogram.getBucketType() instanceof HistogramData.ExponentialBuckets) {
- HistogramData.ExponentialBuckets buckets =
- (HistogramData.ExponentialBuckets) inputHistogram.getBucketType();
- Base2Exponent base2Exp = new Base2Exponent();
- base2Exp.setNumberOfBuckets(numberOfBuckets);
- base2Exp.setScale(buckets.getScale());
- outputHistogram2.setBucketOptions(new BucketOptions().setExponential(base2Exp));
- } else {
- throw new HistogramParsingException(
- "Unable to encode Int64 Histogram, bucket is not recognized");
- }
-
- outputHistogram2.setCount(inputHistogram.getTotalCount());
-
- List<Long> bucketCounts = new ArrayList<>();
-
- Arrays.stream(inputHistogram.getBucketCount())
- .forEach(
- val -> {
- bucketCounts.add(val);
- });
-
- outputHistogram2.setBucketCounts(bucketCounts);
-
- ObjectMapper objectMapper = new ObjectMapper();
- String jsonString = objectMapper.writeValueAsString(outputHistogram2);
-
- return ByteString.copyFromUtf8(jsonString);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- static class HistogramParsingException extends RuntimeException {
- public HistogramParsingException(String message) {
- super(message);
- }
- }
-
- /** Decodes to {@link MonitoringInfoConstants.TypeUrns#PER_WORKER_HISTOGRAM}. */
- public static HistogramData decodeInt64Histogram(ByteString payload) {
- try {
- ObjectMapper objectMapper = new ObjectMapper();
- objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
- JsonNode jsonNode = objectMapper.readTree(payload.toStringUtf8()); // parse afterwards
- DataflowHistogramValue newHist = new DataflowHistogramValue();
- newHist.setCount(jsonNode.get("count").asLong());
-
- List<Long> bucketCounts = new ArrayList<>();
- Iterator<JsonNode> itr = jsonNode.get("bucketCounts").iterator();
- while (itr.hasNext()) {
- Long item = itr.next().asLong();
- bucketCounts.add(item);
- }
- newHist.setBucketCounts(bucketCounts);
-
- if (jsonNode.get("bucketOptions").has("linear")) {
- Linear linear = new Linear();
- JsonNode linearNode = jsonNode.get("bucketOptions").get("linear");
- linear.setNumberOfBuckets(linearNode.get("numberOfBuckets").asInt());
- linear.setWidth(linearNode.get("width").asDouble());
- linear.setStart(linearNode.get("start").asDouble());
- newHist.setBucketOptions(new BucketOptions().setLinear(linear));
- } else if (jsonNode.get("bucketOptions").has("exponential")) {
- Base2Exponent base2Exp = new Base2Exponent();
- JsonNode expNode = jsonNode.get("bucketOptions").get("exponential");
- base2Exp.setNumberOfBuckets(expNode.get("numberOfBuckets").asInt());
- base2Exp.setScale(expNode.get("scale").asInt());
- newHist.setBucketOptions(new BucketOptions().setExponential(base2Exp));
- } else {
- throw new HistogramParsingException(
- "Unable to parse Int64 Histogram, bucket is not recognized");
- }
- return new HistogramData(newHist);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
}
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodingsTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodingsTest.java
index 2d7ba61dbc9..dde180b150d 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodingsTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/MonitoringInfoEncodingsTest.java
@@ -17,43 +17,30 @@
*/
package org.apache.beam.runners.core.metrics;
-import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.HistogramParsingException;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.decodeDoubleCounter;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.decodeInt64Counter;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.decodeInt64Distribution;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.decodeInt64Gauge;
-import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.decodeInt64Histogram;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.decodeStringSet;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.encodeDoubleCounter;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.encodeDoubleDistribution;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.encodeInt64Counter;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.encodeInt64Distribution;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.encodeInt64Gauge;
-import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.encodeInt64Histogram;
import static org.apache.beam.runners.core.metrics.MonitoringInfoEncodings.encodeStringSet;
import static org.junit.Assert.assertEquals;
import java.util.Collections;
-import org.apache.beam.sdk.testing.ExpectedLogs;
-import org.apache.beam.sdk.util.HistogramData;
-import org.apache.beam.vendor.grpc.v1p60p1.com.google.protobuf.ByteString;
+import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.ByteString;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
import org.joda.time.Instant;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
/** Tests for {@link MonitoringInfoEncodings}. */
@RunWith(JUnit4.class)
public class MonitoringInfoEncodingsTest {
- @Rule
- public ExpectedLogs monitoringInfoCodingsExpectedLogs =
- ExpectedLogs.none(MonitoringInfoEncodings.class);
-
- @Rule public ExpectedException thrown = ExpectedException.none();
-
@Test
public void testInt64DistributionEncoding() {
DistributionData data = DistributionData.create(1L, 2L, 3L, 4L);
@@ -118,36 +105,4 @@ public class MonitoringInfoEncodingsTest {
assertEquals(ByteString.copyFrom(new byte[] {0x3f, (byte) 0xf0, 0, 0, 0, 0, 0, 0}), payload);
assertEquals(1.0, decodeDoubleCounter(payload), 0.001);
}
-
- @Test
- public void testHistgramInt64EncodingLinearHist() {
- HistogramData.BucketType buckets = HistogramData.LinearBuckets.of(0, 5, 5);
-
- HistogramData inputHistogram = new HistogramData(buckets);
- inputHistogram.record(5, 10, 15, 20);
- ByteString payload = encodeInt64Histogram(inputHistogram);
-
- assertEquals(inputHistogram, decodeInt64Histogram(payload));
- }
-
- @Test
- public void testHistgramInt64EncodingExpHist() {
- HistogramData.BucketType buckets = HistogramData.ExponentialBuckets.of(1, 10);
- HistogramData inputHistogram = new HistogramData(buckets);
- inputHistogram.record(2, 4, 8, 16, 32);
- ByteString payload = encodeInt64Histogram(inputHistogram);
- assertEquals(inputHistogram, decodeInt64Histogram(payload));
- }
-
- @Test
- public void testHistgramInt64EncodingUnsupportedBucket() {
- thrown.expect(HistogramParsingException.class);
- thrown.expectMessage("Unable to encode Int64 Histogram, bucket is not recognized");
-
- HistogramData.BucketType buckets = HistogramData.UnsupportedBuckets.of();
-
- HistogramData inputHistogram = new HistogramData(buckets);
- inputHistogram.record(2, 4, 8, 16, 32);
- encodeInt64Histogram(inputHistogram);
- }
}
diff --git a/sdks/java/core/build.gradle b/sdks/java/core/build.gradle
index 7423cb7c6b8..a8dfbf42f97 100644
--- a/sdks/java/core/build.gradle
+++ b/sdks/java/core/build.gradle
@@ -75,7 +75,6 @@ dependencies {
permitUnusedDeclared library.java.antlr
permitUsedUndeclared library.java.antlr_runtime
// Required to load constants from the model, e.g. max timestamp for global window
- provided library.java.google_api_services_dataflow
shadow project(path: ":model:pipeline", configuration: "shadow")
shadow project(path: ":model:fn-execution", configuration: "shadow")
shadow project(path: ":model:job-management", configuration: "shadow")
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/HistogramData.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/HistogramData.java
index c1ac4bcfba2..65ccda06be6 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/HistogramData.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/HistogramData.java
@@ -17,7 +17,6 @@
*/
package org.apache.beam.sdk.util;
-import com.google.api.services.dataflow.model.DataflowHistogramValue;
import com.google.auto.value.AutoValue;
import com.google.auto.value.extension.memoized.Memoized;
import java.io.Serializable;
@@ -25,8 +24,6 @@ import java.math.RoundingMode;
import java.util.Arrays;
import java.util.Objects;
import javax.annotation.concurrent.GuardedBy;
-import org.apache.beam.sdk.annotations.Internal;
-import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.math.DoubleMath;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.math.IntMath;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -77,41 +74,6 @@ public class HistogramData implements Serializable {
this.sumOfSquaredDeviations = 0;
}
- /**
- * Create a histogram from DataflowHistogramValue proto.
- *
- * @param histogramProto DataflowHistogramValue proto used to populate stats for the histogram.
- */
- public HistogramData(DataflowHistogramValue histogramProto) {
- int numBuckets;
- if (histogramProto.getBucketOptions().getLinear() != null) {
- double start = histogramProto.getBucketOptions().getLinear().getStart();
- double width = histogramProto.getBucketOptions().getLinear().getWidth();
- numBuckets = histogramProto.getBucketOptions().getLinear().getNumberOfBuckets();
- this.bucketType = LinearBuckets.of(start, width, numBuckets);
- this.buckets = new long[bucketType.getNumBuckets()];
-
- int idx = 0;
- for (long val : histogramProto.getBucketCounts()) {
- this.buckets[idx] = val;
- this.numBoundedBucketRecords += val;
- idx++;
- }
- } else {
- // Assume it's a exponential histogram if its not linear
- int scale = histogramProto.getBucketOptions().getExponential().getScale();
- numBuckets = histogramProto.getBucketOptions().getExponential().getNumberOfBuckets();
- this.bucketType = ExponentialBuckets.of(scale, numBuckets);
- this.buckets = new long[bucketType.getNumBuckets()];
- int idx = 0;
- for (long val : histogramProto.getBucketCounts()) {
- this.buckets[idx] = val;
- this.numBoundedBucketRecords += val;
- idx++;
- }
- }
- }
-
public BucketType getBucketType() {
return this.bucketType;
}
@@ -331,10 +293,6 @@ public class HistogramData implements Serializable {
return numTopRecords;
}
- public synchronized long[] getBucketCount() {
- return buckets;
- }
-
public synchronized double getTopBucketMean() {
return numTopRecords == 0 ? 0 : topRecordsSum / numTopRecords;
}
@@ -615,42 +573,6 @@ public class HistogramData implements Serializable {
// Note: equals() and hashCode() are implemented by the AutoValue.
}
- /** Used for testing unsupported Bucket formats. */
- @AutoValue
- @Internal
- @VisibleForTesting
- public abstract static class UnsupportedBuckets implements BucketType {
-
- public static UnsupportedBuckets of() {
- return new AutoValue_HistogramData_UnsupportedBuckets(0);
- }
-
- @Override
- public int getBucketIndex(double value) {
- return 0;
- }
-
- @Override
- public double getBucketSize(int index) {
- return 0;
- }
-
- @Override
- public double getAccumulatedBucketSize(int index) {
- return 0;
- }
-
- @Override
- public double getRangeFrom() {
- return 0;
- }
-
- @Override
- public double getRangeTo() {
- return 0;
- }
- }
-
@Override
public synchronized boolean equals(@Nullable Object object) {
if (object instanceof HistogramData) {
diff --git a/sdks/java/harness/build.gradle b/sdks/java/harness/build.gradle
index ed3034c0861..2de578cb32c 100644
--- a/sdks/java/harness/build.gradle
+++ b/sdks/java/harness/build.gradle
@@ -30,7 +30,6 @@ dependencies {
provided project(path: ":model:pipeline", configuration: "shadow")
provided project(path: ":sdks:java:core", configuration: "shadow")
provided project(path: ":sdks:java:transform-service:launcher")
- provided library.java.google_api_services_dataflow
provided library.java.avro
provided library.java.jackson_databind
provided library.java.joda_time
@@ -80,5 +79,4 @@ dependencies {
shadowTest project(path: ":sdks:java:core", configuration: "shadowTest")
shadowTestRuntimeClasspath library.java.slf4j_jdk14
permitUnusedDeclared library.java.avro
- permitUnusedDeclared library.java.google_api_services_dataflow
}