You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ro...@apache.org on 2023/11/01 14:30:52 UTC
(beam) branch master updated: [YAML] Java Filter and explode operations. (#29110)
This is an automated email from the ASF dual-hosted git repository.
robertwb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 0d478daf760 [YAML] Java Filter and explode operations. (#29110)
0d478daf760 is described below
commit 0d478daf760b4735bbb4d4718fc404a8e578d1fe
Author: Robert Bradshaw <ro...@gmail.com>
AuthorDate: Wed Nov 1 07:30:43 2023 -0700
[YAML] Java Filter and explode operations. (#29110)
---
.../transforms/providers/ErrorHandling.java | 65 ++++++
.../providers/JavaExplodeTransformProvider.java | 225 +++++++++++++++++++++
.../providers/JavaFilterTransformProvider.java | 181 +++++++++++++++++
.../JavaMapToFieldsTransformProvider.java | 47 ++---
.../schemas/transforms/providers/JavaRowUdf.java | 4 +
.../JavaExplodeTransformProviderTest.java | 104 ++++++++++
.../providers/JavaFilterTransformProviderTest.java | 119 +++++++++++
.../JavaMapToFieldsTransformProviderTest.java | 6 +-
.../apache_beam/yaml/standard_providers.yaml | 14 +-
sdks/python/apache_beam/yaml/yaml_mapping.py | 5 +-
10 files changed, 725 insertions(+), 45 deletions(-)
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/ErrorHandling.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/ErrorHandling.java
new file mode 100644
index 00000000000..3fab157a0e5
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/ErrorHandling.java
@@ -0,0 +1,65 @@
+/*
+ * 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.schemas.transforms.providers;
+
+import com.google.auto.value.AutoValue;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription;
+import org.apache.beam.sdk.values.Row;
+
+@AutoValue
+public abstract class ErrorHandling {
+ @SchemaFieldDescription("The name of the output PCollection containing failed writes.")
+ public abstract String getOutput();
+
+ public static Builder builder() {
+ return new AutoValue_ErrorHandling.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setOutput(String output);
+
+ public abstract ErrorHandling build();
+ }
+
+ public static boolean hasOutput(@Nullable ErrorHandling errorHandling) {
+ return getOutputOrNull(errorHandling) != null;
+ }
+
+ public static @Nullable String getOutputOrNull(@Nullable ErrorHandling errorHandling) {
+ return errorHandling == null ? null : errorHandling.getOutput();
+ }
+
+ public static Schema errorSchema(Schema inputSchema) {
+ return Schema.of(
+ Schema.Field.of("failed_row", Schema.FieldType.row(inputSchema)),
+ Schema.Field.of("error_message", Schema.FieldType.STRING));
+ }
+
+ @SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+ })
+ public static Row errorRecord(Schema errorSchema, Row inputRow, Throwable th) {
+ return Row.withSchema(errorSchema)
+ .withFieldValue("failed_row", inputRow)
+ .withFieldValue("error_message", th.getMessage())
+ .build();
+ }
+}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaExplodeTransformProvider.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaExplodeTransformProvider.java
new file mode 100644
index 00000000000..48ce5e33d9f
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaExplodeTransformProvider.java
@@ -0,0 +1,225 @@
+/*
+ * 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.schemas.transforms.providers;
+
+import com.google.auto.service.AutoService;
+import com.google.auto.value.AutoValue;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
+import org.apache.beam.sdk.schemas.transforms.TypedSchemaTransformProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.Booleans;
+
+/**
+ * An implementation of {@link TypedSchemaTransformProvider} for Explode.
+ *
+ * <p><b>Internal only:</b> This class is actively being worked on, and it will likely change. We
+ * provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
+ * repository.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+@AutoService(SchemaTransformProvider.class)
+public class JavaExplodeTransformProvider
+ extends TypedSchemaTransformProvider<JavaExplodeTransformProvider.Configuration> {
+ protected static final String INPUT_ROWS_TAG = "input";
+ protected static final String OUTPUT_ROWS_TAG = "output";
+
+ @Override
+ protected Class<Configuration> configurationClass() {
+ return Configuration.class;
+ }
+
+ @Override
+ protected SchemaTransform from(Configuration configuration) {
+ return new ExplodeTransform(configuration);
+ }
+
+ @Override
+ public String identifier() {
+ return "beam:schematransform:org.apache.beam:yaml:explode:v1";
+ }
+
+ @Override
+ public List<String> inputCollectionNames() {
+ return Collections.singletonList(INPUT_ROWS_TAG);
+ }
+
+ @Override
+ public List<String> outputCollectionNames() {
+ return Collections.singletonList(OUTPUT_ROWS_TAG);
+ }
+
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class Configuration {
+ @Nullable
+ public abstract List<String> getFields();
+
+ @Nullable
+ public abstract Boolean getCrossProduct();
+
+ public static Builder builder() {
+ return new AutoValue_JavaExplodeTransformProvider_Configuration.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+
+ public abstract Builder setFields(List<String> fields);
+
+ public abstract Builder setCrossProduct(@Nullable Boolean append);
+
+ public abstract Configuration build();
+ }
+ }
+
+ /** A {@link SchemaTransform} for Explode. */
+ protected static class ExplodeTransform extends SchemaTransform {
+
+ private final Configuration configuration;
+
+ ExplodeTransform(Configuration configuration) {
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ Schema inputSchema = input.get(INPUT_ROWS_TAG).getSchema();
+ Schema.Builder outputSchemaBuilder = new Schema.Builder();
+ for (Schema.Field field : inputSchema.getFields()) {
+ if (configuration.getFields().contains(field.getName())) {
+ if (field.getType().getCollectionElementType() == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Exploded field %s must be an iterable type, got %s.",
+ field.getName(), field.getType()));
+ } else {
+ outputSchemaBuilder =
+ outputSchemaBuilder.addField(
+ field.getName(), field.getType().getCollectionElementType());
+ }
+ } else {
+ outputSchemaBuilder = outputSchemaBuilder.addField(field);
+ }
+ }
+ Schema outputSchema = outputSchemaBuilder.build();
+
+ PCollection<Row> result =
+ input
+ .get(INPUT_ROWS_TAG)
+ .apply(
+ "Explode",
+ ParDo.of(
+ createDoFn(
+ configuration.getFields(),
+ configuration.getCrossProduct(),
+ outputSchema)));
+ result.setRowSchema(outputSchema);
+
+ return PCollectionRowTuple.of(OUTPUT_ROWS_TAG, result);
+ }
+
+ private static DoFn<Row, Row> createDoFn(
+ List<String> fields, Boolean crossProductObj, Schema outputSchema) {
+ boolean crossProduct;
+ if (crossProductObj == null) {
+ if (fields.size() > 1) {
+ throw new IllegalArgumentException(
+ "boolean cross product parameter required to explode more than one field");
+ }
+ crossProduct = false;
+ } else {
+ crossProduct = crossProductObj;
+ }
+ int numFields = outputSchema.getFields().size();
+ boolean[] toExplode =
+ Booleans.toArray(
+ IntStream.range(0, numFields)
+ .mapToObj(index -> fields.contains(outputSchema.getField(index).getName()))
+ .collect(Collectors.toList()));
+ if (crossProduct) {
+ return new DoFn<Row, Row>() {
+ @ProcessElement
+ public void processElement(@Element Row inputRow, OutputReceiver<Row> out) {
+ emitCrossProduct(inputRow, 0, new Object[numFields], out);
+ }
+
+ private void emitCrossProduct(
+ Row inputRow, int index, Object[] current, OutputReceiver<Row> out) {
+ if (index == numFields) {
+ out.output(Row.withSchema(outputSchema).attachValues(ImmutableList.copyOf(current)));
+ } else if (toExplode[index]) {
+ for (Object value : inputRow.getIterable(index)) {
+ current[index] = value;
+ emitCrossProduct(inputRow, index + 1, current, out);
+ }
+ } else {
+ current[index] = inputRow.getValue(index);
+ emitCrossProduct(inputRow, index + 1, current, out);
+ }
+ }
+ };
+ } else {
+ return new DoFn<Row, Row>() {
+ @ProcessElement
+ public void processElement(@Element Row inputRow, OutputReceiver<Row> out) {
+ @SuppressWarnings("rawtypes")
+ Iterator[] iterators = new Iterator[numFields];
+ for (int i = 0; i < numFields; i++) {
+ if (toExplode[i]) {
+ iterators[i] = inputRow.getIterable(i).iterator();
+ }
+ }
+ while (IntStream.range(0, numFields)
+ .anyMatch(index -> toExplode[index] && iterators[index].hasNext())) {
+ Row.Builder builder = Row.withSchema(outputSchema);
+ for (int i = 0; i < numFields; i++) {
+ if (toExplode[i]) {
+ if (iterators[i].hasNext()) {
+ builder.addValue(iterators[i].next());
+ } else {
+ builder.addValue(null);
+ }
+ } else {
+ builder.addValue(inputRow.getValue(i));
+ }
+ }
+ out.output(builder.build());
+ }
+ }
+ };
+ }
+ }
+ }
+}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaFilterTransformProvider.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaFilterTransformProvider.java
new file mode 100644
index 00000000000..4ae8d2e41b3
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaFilterTransformProvider.java
@@ -0,0 +1,181 @@
+/*
+ * 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.schemas.transforms.providers;
+
+import com.google.auto.service.AutoService;
+import com.google.auto.value.AutoValue;
+import java.net.MalformedURLException;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
+import org.apache.beam.sdk.schemas.transforms.TypedSchemaTransformProvider;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+
+/**
+ * An implementation of {@link TypedSchemaTransformProvider} for Filter for the java language.
+ *
+ * <p><b>Internal only:</b> This class is actively being worked on, and it will likely change. We
+ * provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
+ * repository.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+@AutoService(SchemaTransformProvider.class)
+public class JavaFilterTransformProvider
+ extends TypedSchemaTransformProvider<JavaFilterTransformProvider.Configuration> {
+ protected static final String INPUT_ROWS_TAG = "input";
+ protected static final String OUTPUT_ROWS_TAG = "output";
+
+ @Override
+ protected Class<Configuration> configurationClass() {
+ return Configuration.class;
+ }
+
+ @Override
+ protected SchemaTransform from(Configuration configuration) {
+ return new JavaFilterTransform(configuration);
+ }
+
+ @Override
+ public String identifier() {
+ return "beam:schematransform:org.apache.beam:yaml:filter-java:v1";
+ }
+
+ @Override
+ public List<String> inputCollectionNames() {
+ return Collections.singletonList(INPUT_ROWS_TAG);
+ }
+
+ @Override
+ public List<String> outputCollectionNames() {
+ return Collections.singletonList(OUTPUT_ROWS_TAG);
+ }
+
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class Configuration {
+ @Nullable
+ public abstract String getLanguage();
+
+ public abstract JavaRowUdf.Configuration getKeep();
+
+ @Nullable
+ public abstract ErrorHandling getErrorHandling();
+
+ public static Builder builder() {
+ return new AutoValue_JavaFilterTransformProvider_Configuration.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+
+ public abstract Builder setLanguage(String language);
+
+ public abstract Builder setKeep(JavaRowUdf.Configuration keep);
+
+ public abstract Builder setErrorHandling(ErrorHandling errorHandling);
+
+ public abstract Configuration build();
+ }
+ }
+
+ /** A {@link SchemaTransform} for Filter-java. */
+ protected static class JavaFilterTransform extends SchemaTransform {
+
+ private final Configuration configuration;
+
+ JavaFilterTransform(Configuration configuration) {
+ this.configuration = configuration;
+ }
+
+ @Override
+ public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ Schema inputSchema = input.get(INPUT_ROWS_TAG).getSchema();
+ JavaRowUdf keepFn;
+ try {
+ keepFn = new JavaRowUdf(this.configuration.getKeep(), inputSchema);
+ } catch (MalformedURLException
+ | ReflectiveOperationException
+ | StringCompiler.CompileException exn) {
+ throw new RuntimeException(exn);
+ }
+ if (!keepFn.getOutputType().withNullable(false).equals(Schema.FieldType.BOOLEAN)) {
+ throw new RuntimeException(
+ String.format(
+ "KeepFn %s must return a boolean, but returns %s instead.",
+ this.configuration.getKeep(), keepFn.getOutputType()));
+ }
+ boolean handleErrors = ErrorHandling.hasOutput(configuration.getErrorHandling());
+ Schema errorSchema = ErrorHandling.errorSchema(inputSchema);
+
+ PCollectionTuple pcolls =
+ input
+ .get(INPUT_ROWS_TAG)
+ .apply(
+ "Filter",
+ ParDo.of(createDoFn(keepFn, errorSchema, handleErrors))
+ .withOutputTags(filteredValues, TupleTagList.of(errorValues)));
+ pcolls.get(filteredValues).setRowSchema(inputSchema);
+ pcolls.get(errorValues).setRowSchema(errorSchema);
+
+ PCollectionRowTuple result =
+ PCollectionRowTuple.of(OUTPUT_ROWS_TAG, pcolls.get(filteredValues));
+ if (handleErrors) {
+ result = result.and(configuration.getErrorHandling().getOutput(), pcolls.get(errorValues));
+ }
+ return result;
+ }
+
+ private static final TupleTag<Row> filteredValues = new TupleTag<Row>() {};
+ private static final TupleTag<Row> errorValues = new TupleTag<Row>() {};
+
+ private static DoFn<Row, Row> createDoFn(
+ JavaRowUdf keepFn, Schema errorSchema, boolean handleErrors) {
+ return new DoFn<Row, Row>() {
+ @ProcessElement
+ public void processElement(@Element Row inputRow, MultiOutputReceiver out) {
+ boolean keep = false;
+ try {
+ keep = (boolean) keepFn.getFunction().apply(inputRow);
+ } catch (Exception exn) {
+ if (handleErrors) {
+ out.get(errorValues).output(ErrorHandling.errorRecord(errorSchema, inputRow, exn));
+ } else {
+ throw new RuntimeException(exn);
+ }
+ }
+ if (keep) {
+ out.get(filteredValues).output(inputRow);
+ }
+ }
+ };
+ }
+ }
+}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaMapToFieldsTransformProvider.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaMapToFieldsTransformProvider.java
index ddf892f03fc..2e2042aef05 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaMapToFieldsTransformProvider.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaMapToFieldsTransformProvider.java
@@ -28,7 +28,6 @@ import javax.annotation.Nullable;
import org.apache.beam.sdk.schemas.AutoValueSchema;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
-import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription;
import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
import org.apache.beam.sdk.schemas.transforms.TypedSchemaTransformProvider;
@@ -68,7 +67,7 @@ public class JavaMapToFieldsTransformProvider
@Override
public String identifier() {
- return String.format("beam:schematransform:org.apache.beam:yaml:map_to_fields-java:v1");
+ return "beam:schematransform:org.apache.beam:yaml:map_to_fields-java:v1";
}
@Override
@@ -117,23 +116,6 @@ public class JavaMapToFieldsTransformProvider
public abstract Configuration build();
}
-
- @AutoValue
- public abstract static class ErrorHandling {
- @SchemaFieldDescription("The name of the output PCollection containing failed writes.")
- public abstract String getOutput();
-
- public static Builder builder() {
- return new AutoValue_JavaMapToFieldsTransformProvider_Configuration_ErrorHandling.Builder();
- }
-
- @AutoValue.Builder
- public abstract static class Builder {
- public abstract Builder setOutput(String output);
-
- public abstract ErrorHandling build();
- }
- }
}
/** A {@link SchemaTransform} for MapToFields-java. */
@@ -192,13 +174,8 @@ public class JavaMapToFieldsTransformProvider
}
}
Schema outputSchema = outputSchemaBuilder.build();
- boolean handleErrors =
- configuration.getErrorHandling() != null
- && configuration.getErrorHandling().getOutput() != null;
- Schema errorSchema =
- Schema.of(
- Schema.Field.of("failed_row", Schema.FieldType.row(inputSchema)),
- Schema.Field.of("error_message", Schema.FieldType.STRING));
+ boolean handleErrors = ErrorHandling.hasOutput(configuration.getErrorHandling());
+ Schema errorSchema = ErrorHandling.errorSchema(inputSchema);
PCollectionTuple pcolls =
input
@@ -226,24 +203,24 @@ public class JavaMapToFieldsTransformProvider
return new DoFn<Row, Row>() {
@ProcessElement
public void processElement(@Element Row inputRow, MultiOutputReceiver out) {
+ Row outputRow;
try {
- Row.Builder outputRow = Row.withSchema(outputSchema);
+ Row.Builder builder = Row.withSchema(outputSchema);
for (JavaRowUdf udf : udfs) {
- outputRow.addValue(udf.getFunction().apply(inputRow));
+ builder.addValue(udf.getFunction().apply(inputRow));
}
- out.get(mappedValues).output(outputRow.build());
+ outputRow = builder.build();
} catch (Exception exn) {
if (handleErrors) {
- out.get(errorValues)
- .output(
- Row.withSchema(errorSchema)
- .withFieldValue("failed_row", inputRow)
- .withFieldValue("error_message", exn.getMessage())
- .build());
+ out.get(errorValues).output(ErrorHandling.errorRecord(errorSchema, inputRow, exn));
+ outputRow = null;
} else {
throw new RuntimeException(exn);
}
}
+ if (outputRow != null) {
+ out.get(mappedValues).output(outputRow);
+ }
}
};
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaRowUdf.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaRowUdf.java
index e18df5b29a5..2ec0a9a60cd 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaRowUdf.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaRowUdf.java
@@ -313,6 +313,8 @@ public class JavaRowUdf implements Serializable {
.put(Schema.TypeName.FLOAT, Float.class)
.put(Schema.TypeName.DOUBLE, Double.class)
.put(Schema.TypeName.BOOLEAN, Boolean.class)
+ .put(Schema.TypeName.BYTES, byte[].class)
+ .put(Schema.TypeName.STRING, String.class)
.put(Schema.TypeName.DECIMAL, BigDecimal.class)
.build();
@@ -325,6 +327,8 @@ public class JavaRowUdf implements Serializable {
.put(Schema.TypeName.FLOAT, float.class)
.put(Schema.TypeName.DOUBLE, double.class)
.put(Schema.TypeName.BOOLEAN, boolean.class)
+ .put(Schema.TypeName.BYTES, byte[].class)
+ .put(Schema.TypeName.STRING, String.class)
.put(Schema.TypeName.DECIMAL, BigDecimal.class)
.build();
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaExplodeTransformProviderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaExplodeTransformProviderTest.java
new file mode 100644
index 00000000000..7c8166d92b9
--- /dev/null
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaExplodeTransformProviderTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.schemas.transforms.providers;
+
+import java.util.List;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+public class JavaExplodeTransformProviderTest {
+ @Rule public TestPipeline pipeline = TestPipeline.create();
+
+ private static final Schema INPUT_SCHEMA =
+ Schema.of(
+ Schema.Field.of("a", Schema.FieldType.iterable(Schema.FieldType.INT32)),
+ Schema.Field.of("b", Schema.FieldType.DOUBLE),
+ Schema.Field.of("c", Schema.FieldType.array(Schema.FieldType.STRING)));
+
+ private static final Schema OUTPUT_SCHEMA =
+ Schema.of(
+ Schema.Field.of("a", Schema.FieldType.INT32),
+ Schema.Field.of("b", Schema.FieldType.DOUBLE),
+ Schema.Field.of("c", Schema.FieldType.STRING));
+
+ private static final List<Row> INPUT_ROWS =
+ ImmutableList.of(
+ Row.withSchema(INPUT_SCHEMA)
+ .addValues(ImmutableList.of(1, 2), 1.5, ImmutableList.of("x", "y"))
+ .build());
+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testCrossProduct() {
+ PCollection<Row> input = pipeline.apply(Create.of(INPUT_ROWS)).setRowSchema(INPUT_SCHEMA);
+
+ PCollection<Row> exploded =
+ PCollectionRowTuple.of(JavaExplodeTransformProvider.INPUT_ROWS_TAG, input)
+ .apply(
+ new JavaExplodeTransformProvider()
+ .from(
+ JavaExplodeTransformProvider.Configuration.builder()
+ .setFields(ImmutableList.of("a", "c"))
+ .setCrossProduct(true)
+ .build()))
+ .get(JavaExplodeTransformProvider.OUTPUT_ROWS_TAG);
+
+ PAssert.that(exploded)
+ .containsInAnyOrder(
+ Row.withSchema(OUTPUT_SCHEMA).addValues(1, 1.5, "x").build(),
+ Row.withSchema(OUTPUT_SCHEMA).addValues(2, 1.5, "x").build(),
+ Row.withSchema(OUTPUT_SCHEMA).addValues(1, 1.5, "y").build(),
+ Row.withSchema(OUTPUT_SCHEMA).addValues(2, 1.5, "y").build());
+
+ pipeline.run();
+ }
+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testZipProduct() {
+ PCollection<Row> input = pipeline.apply(Create.of(INPUT_ROWS)).setRowSchema(INPUT_SCHEMA);
+
+ PCollection<Row> exploded =
+ PCollectionRowTuple.of(JavaExplodeTransformProvider.INPUT_ROWS_TAG, input)
+ .apply(
+ new JavaExplodeTransformProvider()
+ .from(
+ JavaExplodeTransformProvider.Configuration.builder()
+ .setFields(ImmutableList.of("a", "c"))
+ .setCrossProduct(false)
+ .build()))
+ .get(JavaExplodeTransformProvider.OUTPUT_ROWS_TAG);
+
+ PAssert.that(exploded)
+ .containsInAnyOrder(
+ Row.withSchema(OUTPUT_SCHEMA).addValues(1, 1.5, "x").build(),
+ Row.withSchema(OUTPUT_SCHEMA).addValues(2, 1.5, "y").build());
+
+ pipeline.run();
+ }
+}
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaFilterTransformProviderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaFilterTransformProviderTest.java
new file mode 100644
index 00000000000..b269b12f155
--- /dev/null
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaFilterTransformProviderTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.schemas.transforms.providers;
+
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+public class JavaFilterTransformProviderTest {
+ @Rule public TestPipeline pipeline = TestPipeline.create();
+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testFilter() {
+ Schema inputSchema =
+ Schema.of(
+ Schema.Field.of("a", Schema.FieldType.STRING),
+ Schema.Field.of("b", Schema.FieldType.INT32),
+ Schema.Field.of("c", Schema.FieldType.DOUBLE));
+
+ PCollection<Row> input =
+ pipeline
+ .apply(
+ Create.of(
+ Row.withSchema(inputSchema).addValues("foo", 2, 0.5).build(),
+ Row.withSchema(inputSchema).addValues("bar", 4, 0.25).build()))
+ .setRowSchema(inputSchema);
+
+ PCollection<Row> renamed =
+ PCollectionRowTuple.of(JavaFilterTransformProvider.INPUT_ROWS_TAG, input)
+ .apply(
+ new JavaFilterTransformProvider()
+ .from(
+ JavaFilterTransformProvider.Configuration.builder()
+ .setKeep(
+ JavaRowUdf.Configuration.builder()
+ .setExpression("b + c > 3")
+ .build())
+ .build()))
+ .get(JavaFilterTransformProvider.OUTPUT_ROWS_TAG);
+
+ PAssert.that(renamed)
+ .containsInAnyOrder(
+ Row.withSchema(inputSchema)
+ .withFieldValue("a", "bar")
+ .withFieldValue("b", 4)
+ .withFieldValue("c", 0.25)
+ .build());
+
+ pipeline.run();
+ }
+
+ @Test
+ @Category(NeedsRunner.class)
+ public void testErrorHandling() {
+ Schema inputSchema = Schema.of(Schema.Field.of("s", Schema.FieldType.STRING));
+
+ PCollection<Row> input =
+ pipeline
+ .apply(
+ Create.of(
+ Row.withSchema(inputSchema).addValues("short").build(),
+ Row.withSchema(inputSchema).addValues("looooooooooooong").build()))
+ .setRowSchema(inputSchema);
+
+ PCollectionRowTuple result =
+ PCollectionRowTuple.of(JavaFilterTransformProvider.INPUT_ROWS_TAG, input)
+ .apply(
+ new JavaFilterTransformProvider()
+ .from(
+ JavaFilterTransformProvider.Configuration.builder()
+ .setLanguage("java")
+ .setKeep(
+ JavaRowUdf.Configuration.builder()
+ .setExpression("s.charAt(7) == 'o'")
+ .build())
+ .setErrorHandling(ErrorHandling.builder().setOutput("errors").build())
+ .build()));
+
+ PCollection<Row> good = result.get(JavaFilterTransformProvider.OUTPUT_ROWS_TAG);
+ PAssert.that(good)
+ .containsInAnyOrder(
+ Row.withSchema(inputSchema).withFieldValue("s", "looooooooooooong").build());
+
+ PCollection<Row> errors = result.get("errors");
+ Schema errorSchema = errors.getSchema();
+ PAssert.that(errors)
+ .containsInAnyOrder(
+ Row.withSchema(errorSchema)
+ .withFieldValue(
+ "failed_row", Row.withSchema(inputSchema).addValues("short").build())
+ .withFieldValue("error_message", "String index out of range: 7")
+ .build());
+ pipeline.run();
+ }
+}
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaMapToFieldsTransformProviderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaMapToFieldsTransformProviderTest.java
index 64fc48564cc..6ad6f353a4d 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaMapToFieldsTransformProviderTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/transforms/providers/JavaMapToFieldsTransformProviderTest.java
@@ -171,11 +171,7 @@ public class JavaMapToFieldsTransformProviderTest {
+ " }"
+ "}")
.build()))
- .setErrorHandling(
- JavaMapToFieldsTransformProvider.Configuration.ErrorHandling
- .builder()
- .setOutput("errors")
- .build())
+ .setErrorHandling(ErrorHandling.builder().setOutput("errors").build())
.build()));
PCollection<Row> sqrts = result.get(JavaMapToFieldsTransformProvider.OUTPUT_ROWS_TAG);
diff --git a/sdks/python/apache_beam/yaml/standard_providers.yaml b/sdks/python/apache_beam/yaml/standard_providers.yaml
index 01049569ede..c612d441208 100644
--- a/sdks/python/apache_beam/yaml/standard_providers.yaml
+++ b/sdks/python/apache_beam/yaml/standard_providers.yaml
@@ -30,7 +30,9 @@
- type: renaming
transforms:
'MapToFields-java': 'MapToFields-java'
- 'MapToFields-generic': 'MapToFields-generic'
+ 'MapToFields-generic': 'MapToFields-java'
+ 'Filter-java': 'Filter-java'
+ 'Explode': 'Explode'
config:
mappings:
'MapToFields-generic':
@@ -45,10 +47,18 @@
drop: 'drop'
fields: 'fields'
error_handling: 'errorHandling'
+ 'Filter-java':
+ language: 'language'
+ keep: 'keep'
+ error_handling: 'errorHandling'
+ 'Explode':
+ fields: 'fields'
+ cross_product: 'crossProduct'
underlying_provider:
type: beamJar
transforms:
MapToFields-java: "beam:schematransform:org.apache.beam:yaml:map_to_fields-java:v1"
- MapToFields-generic: "beam:schematransform:org.apache.beam:yaml:map_to_fields-java:v1"
+ Filter-java: "beam:schematransform:org.apache.beam:yaml:filter-java:v1"
+ Explode: "beam:schematransform:org.apache.beam:yaml:explode:v1"
config:
gradle_target: 'sdks:java:extensions:sql:expansion-service:shadowJar'
diff --git a/sdks/python/apache_beam/yaml/yaml_mapping.py b/sdks/python/apache_beam/yaml/yaml_mapping.py
index 501c7a5c57b..3497660aedc 100644
--- a/sdks/python/apache_beam/yaml/yaml_mapping.py
+++ b/sdks/python/apache_beam/yaml/yaml_mapping.py
@@ -239,8 +239,6 @@ def maybe_with_exception_handling_transform_fn(transform_fn):
return expand
-# TODO(yaml): This should be available in all environments, in which case
-# we choose the one that matches best.
class _Explode(beam.PTransform):
def __init__(
self,
@@ -289,11 +287,12 @@ class _Explode(beam.PTransform):
copy[field] = values[ix]
yield beam.Row(**copy)
+ cross_product = self._cross_product
return (
pcoll
| beam.FlatMap(
lambda row:
- (explode_cross_product if self._cross_product else explode_zip)
+ (explode_cross_product if cross_product else explode_zip)
({name: getattr(row, name)
for name in all_fields}, to_explode)))