You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ec...@apache.org on 2019/09/05 14:47:53 UTC
[beam] branch spark-runner_structured-streaming updated (c350188 ->
cfdf4a4)
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a change to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git.
from c350188 Use "sparkMaster" in local mode to obtain number of shuffle partitions + spotless apply
new 031754c Wrap Beam Coders into Spark Encoders using ExpressionEncoder: serialization part
new d613d6b Wrap Beam Coders into Spark Encoders using ExpressionEncoder: deserialization part
new 84f2cbd9 Fix EncoderHelpers.fromBeamCoder() visibility
new 95fd272 type erasure: spark encoders require a Class<T>, pass Object and cast to Class<T>
new a5d49f5 Fix scala Product in Encoders to avoid StackEverflow
new 23735e4 Conform to spark ExpressionEncoders: pass classTags, implement scala Product, pass children from within the ExpressionEncoder
new e96af88 Fix visibility of serializer and deserializer
new 2bf4cd9 Add a simple spark native test to test Beam coders wrapping into Spark Encoders
new b11e100 Fix code generation in Beam coder wrapper
new a758985 Lazy init coder because coder instance cannot be interpolated by catalyst
new 2c94eef Fix call to scala Fucntion1 in coder lazy init
new cbd7c2b Fix warning in coder construction by reflexion
new 668227b Fix ExpressionEncoder generated code: typos, try catch, fqcn
new 959664f Fix getting the output value in code generation
new 8bbf991 Fix beam coder lazy init using reflexion
new 723c004 Add try catch around reflexion call in lazy init of beam coder
new 91e923c Cast coder instanciated by reflection
new c4a4464 Remove lazy init of beam coder because there is no generic way on instanciating a beam coder
new 4351304 Remove example code
new ef69410 Fix equal and hashcode
new d8b8b42 Fix generated code: uniform exceptions catching, fix parenthesis and variable declarations
new 0fe6f9b Add an assert of equality in the encoders test
new b879123 Apply spotless and checkstyle and add javadocs
new cfdf4a4 Improve exceptions catching
The 24 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.
Summary of changes:
runners/spark/build.gradle | 1 +
.../translation/helpers/EncoderHelpers.java | 290 +++++++++++++++++++++
.../structuredstreaming/utils/EncodersTest.java} | 38 +--
3 files changed, 313 insertions(+), 16 deletions(-)
copy runners/{google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/apiary/ApiaryTest.java => spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java} (51%)
[beam] 24/24: Improve exceptions catching
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit cfdf4a4ee43091a281d6dcff021f290d698ed2a7
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 15:53:10 2019 +0200
Improve exceptions catching
---
.../translation/helpers/EncoderHelpers.java | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index f990121..f4ea6fa 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -147,8 +147,8 @@ public class EncoderHelpers {
$beamCoder.encode(${input.value}, baos);
${ev.value} = baos.toByteArray();
}
- } catch (Exception e) {
- throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
+ } catch (java.io.IOException e) {
+ throw org.apache.beam.sdk.util.UserCodeException.wrap(e);
}
*/
List<String> parts = new ArrayList<>();
@@ -160,7 +160,7 @@ public class EncoderHelpers {
parts.add(".encode(");
parts.add(", baos); ");
parts.add(
- " = baos.toByteArray();}} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
+ " = baos.toByteArray();}} catch (java.io.IOException e) {throw org.apache.beam.sdk.util.UserCodeException.wrap(e);}");
StringContext sc =
new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
@@ -262,8 +262,8 @@ public class EncoderHelpers {
${input.isNull} ?
${CodeGenerator.defaultValue(dataType)} :
($javaType) $beamCoder.decode(new java.io.ByteArrayInputStream(${input.value}));
- } catch (Exception e) {
- throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
+ } catch (java.io.IOException e) {
+ throw org.apache.beam.sdk.util.UserCodeException.wrap(e);
}
*/
@@ -277,7 +277,7 @@ public class EncoderHelpers {
parts.add(") ");
parts.add(".decode(new java.io.ByteArrayInputStream(");
parts.add(
- ")); } catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
+ ")); } catch (java.io.IOException e) {throw org.apache.beam.sdk.util.UserCodeException.wrap(e);}");
StringContext sc =
new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
[beam] 20/24: Fix equal and hashcode
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit ef694106cf61a66e823b5e7b5a233629af83d029
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 14:49:37 2019 +0200
Fix equal and hashcode
---
.../structuredstreaming/translation/helpers/EncoderHelpers.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 83243b3..91aaaf9 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -195,11 +195,11 @@ public class EncoderHelpers {
return false;
}
EncodeUsingBeamCoder<?> that = (EncodeUsingBeamCoder<?>) o;
- return beamCoder.equals(that.beamCoder);
+ return beamCoder.equals(that.beamCoder) && child.equals(that.child);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), beamCoder);
+ return Objects.hash(super.hashCode(), child, beamCoder);
}
}
@@ -306,11 +306,11 @@ public class EncoderHelpers {
return false;
}
DecodeUsingBeamCoder<?> that = (DecodeUsingBeamCoder<?>) o;
- return classTag.equals(that.classTag) && beamCoder.equals(that.beamCoder);
+ return child.equals(that.child) && classTag.equals(that.classTag) && beamCoder.equals(that.beamCoder);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), classTag, beamCoder);
+ return Objects.hash(super.hashCode(), child, classTag, beamCoder);
}
}
}
[beam] 10/24: Lazy init coder because coder instance cannot be
interpolated by catalyst
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a758985ad24183bbdba3c11ea51b7020bc0865cf
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon Sep 2 17:55:24 2019 +0200
Lazy init coder because coder instance cannot be interpolated by catalyst
---
.../translation/helpers/EncoderHelpers.java | 61 +++++++++++++++-------
.../structuredstreaming/utils/EncodersTest.java | 3 +-
2 files changed, 45 insertions(+), 19 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index cc862cd..f7706cc 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -45,7 +45,6 @@ import org.apache.spark.sql.catalyst.expressions.codegen.ExprCode;
import org.apache.spark.sql.catalyst.expressions.codegen.VariableValue;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.ObjectType;
-import scala.Function1;
import scala.StringContext;
import scala.Tuple2;
import scala.collection.JavaConversions;
@@ -94,17 +93,17 @@ public class EncoderHelpers {
*/
/** A way to construct encoders using generic serializers. */
- public static <T> Encoder<T> fromBeamCoder(Coder<T> coder/*, Class<T> claz*/){
+ public static <T> Encoder<T> fromBeamCoder(Class<? extends Coder<T>> coderClass/*, Class<T> claz*/){
List<Expression> serialiserList = new ArrayList<>();
Class<T> claz = (Class<T>) Object.class;
- serialiserList.add(new EncodeUsingBeamCoder<>(new BoundReference(0, new ObjectType(claz), true), coder));
+ serialiserList.add(new EncodeUsingBeamCoder<>(new BoundReference(0, new ObjectType(claz), true), (Class<Coder<T>>)coderClass));
ClassTag<T> classTag = ClassTag$.MODULE$.apply(claz);
return new ExpressionEncoder<>(
SchemaHelpers.binarySchema(),
false,
JavaConversions.collectionAsScalaIterable(serialiserList).toSeq(),
- new DecodeUsingBeamCoder<>(new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, coder),
+ new DecodeUsingBeamCoder<>(new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, (Class<Coder<T>>)coderClass),
classTag);
/*
@@ -127,11 +126,11 @@ public class EncoderHelpers {
public static class EncodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
private Expression child;
- private Coder<T> beamCoder;
+ private Class<Coder<T>> coderClass;
- public EncodeUsingBeamCoder(Expression child, Coder<T> beamCoder) {
+ public EncodeUsingBeamCoder(Expression child, Class<Coder<T>> coderClass) {
this.child = child;
- this.beamCoder = beamCoder;
+ this.coderClass = coderClass;
}
@Override public Expression child() {
@@ -140,6 +139,7 @@ public class EncoderHelpers {
@Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
// Code to serialize.
+ String beamCoder = lazyInitBeamCoder(ctx, coderClass);
ExprCode input = child.genCode(ctx);
/*
@@ -170,6 +170,7 @@ public class EncoderHelpers {
new VariableValue("output", Array.class));
}
+
@Override public DataType dataType() {
return BinaryType;
}
@@ -179,7 +180,7 @@ public class EncoderHelpers {
case 0:
return child;
case 1:
- return beamCoder;
+ return coderClass;
default:
throw new ArrayIndexOutOfBoundsException("productElement out of bounds");
}
@@ -201,11 +202,11 @@ public class EncoderHelpers {
return false;
}
EncodeUsingBeamCoder<?> that = (EncodeUsingBeamCoder<?>) o;
- return beamCoder.equals(that.beamCoder);
+ return coderClass.equals(that.coderClass);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), beamCoder);
+ return Objects.hash(super.hashCode(), coderClass);
}
}
@@ -237,12 +238,12 @@ public class EncoderHelpers {
private Expression child;
private ClassTag<T> classTag;
- private Coder<T> beamCoder;
+ private Class<Coder<T>> coderClass;
- public DecodeUsingBeamCoder(Expression child, ClassTag<T> classTag, Coder<T> beamCoder) {
+ public DecodeUsingBeamCoder(Expression child, ClassTag<T> classTag, Class<Coder<T>> coderClass) {
this.child = child;
this.classTag = classTag;
- this.beamCoder = beamCoder;
+ this.coderClass = coderClass;
}
@Override public Expression child() {
@@ -251,6 +252,7 @@ public class EncoderHelpers {
@Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
// Code to deserialize.
+ String beamCoder = lazyInitBeamCoder(ctx, coderClass);
ExprCode input = child.genCode(ctx);
String javaType = CodeGenerator.javaType(dataType());
@@ -291,9 +293,10 @@ public class EncoderHelpers {
@Override public Object nullSafeEval(Object input) {
try {
+ Coder<T> beamCoder = coderClass.newInstance();
return beamCoder.decode(new ByteArrayInputStream((byte[]) input));
- } catch (IOException e) {
- throw new IllegalStateException("Error decoding bytes for coder: " + beamCoder, e);
+ } catch (Exception e) {
+ throw new IllegalStateException("Error decoding bytes for coder: " + coderClass, e);
}
}
@@ -308,7 +311,7 @@ public class EncoderHelpers {
case 1:
return classTag;
case 2:
- return beamCoder;
+ return coderClass;
default:
throw new ArrayIndexOutOfBoundsException("productElement out of bounds");
}
@@ -330,11 +333,11 @@ public class EncoderHelpers {
return false;
}
DecodeUsingBeamCoder<?> that = (DecodeUsingBeamCoder<?>) o;
- return classTag.equals(that.classTag) && beamCoder.equals(that.beamCoder);
+ return classTag.equals(that.classTag) && coderClass.equals(that.coderClass);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), classTag, beamCoder);
+ return Objects.hash(super.hashCode(), classTag, coderClass);
}
}
/*
@@ -365,4 +368,26 @@ case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: B
}
*/
+ private static <T> String lazyInitBeamCoder(CodegenContext ctx, Class<Coder<T>> coderClass) {
+ String beamCoderInstance = "beamCoder";
+ ctx.addImmutableStateIfNotExists(coderClass.getName(), beamCoderInstance, v -> {
+ /*
+ CODE GENERATED
+ v = (coderClass) coderClass.newInstance();
+ */
+ List<String> parts = new ArrayList<>();
+ parts.add("");
+ parts.add(" = (");
+ parts.add(") ");
+ parts.add(".newInstance();");
+ StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
+ List<Object> args = new ArrayList<>();
+ args.add(v);
+ args.add(coderClass.getName());
+ args.add(coderClass.getName());
+ return sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
+ });
+ return beamCoderInstance;
+ }
+
}
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
index 7078b0c..0e38fe1 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
@@ -3,6 +3,7 @@ package org.apache.beam.runners.spark.structuredstreaming.utils;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
+import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.spark.sql.SparkSession;
import org.junit.Test;
@@ -23,7 +24,7 @@ public class EncodersTest {
data.add(1);
data.add(2);
data.add(3);
- sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
+ sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.class));
// sparkSession.createDataset(data, EncoderHelpers.genericEncoder());
}
}
[beam] 08/24: Add a simple spark native test to test Beam coders
wrapping into Spark Encoders
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 2bf4cd9bac0e0dfa03d3499f5d84b0d51781d8ba
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Fri Aug 30 17:34:13 2019 +0200
Add a simple spark native test to test Beam coders wrapping into Spark Encoders
---
.../structuredstreaming/utils/EncodersTest.java | 29 ++++++++++++++++++++++
1 file changed, 29 insertions(+)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
new file mode 100644
index 0000000..490e3dc
--- /dev/null
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
@@ -0,0 +1,29 @@
+package org.apache.beam.runners.spark.structuredstreaming.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.spark.sql.SparkSession;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+/**
+ * Test of the wrapping of Beam Coders as Spark ExpressionEncoders.
+ */
+public class EncodersTest {
+
+ @Test
+ public void beamCoderToSparkEncoderTest() {
+ SparkSession sparkSession = SparkSession.builder().appName("beamCoderToSparkEncoderTest")
+ .master("local[4]").getOrCreate();
+ List<Integer> data = new ArrayList<>();
+ data.add(1);
+ data.add(2);
+ data.add(3);
+// sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
+ sparkSession.createDataset(data, EncoderHelpers.genericEncoder());
+ }
+}
[beam] 12/24: Fix warning in coder construction by reflexion
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit cbd7c2bdd7ad96317b4a7545a28d35f467b05795
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed Sep 4 14:55:32 2019 +0200
Fix warning in coder construction by reflexion
---
.../structuredstreaming/translation/helpers/EncoderHelpers.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 694bc24..1d89101 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -293,7 +293,7 @@ public class EncoderHelpers {
@Override public Object nullSafeEval(Object input) {
try {
- Coder<T> beamCoder = coderClass.newInstance();
+ Coder<T> beamCoder = coderClass.getDeclaredConstructor().newInstance();
return beamCoder.decode(new ByteArrayInputStream((byte[]) input));
} catch (Exception e) {
throw new IllegalStateException("Error decoding bytes for coder: " + coderClass, e);
@@ -373,13 +373,13 @@ case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: B
ctx.addImmutableStateIfNotExists(coderClass.getName(), beamCoderInstance, func(v1 -> {
/*
CODE GENERATED
- v = (coderClass) coderClass.newInstance();
+ v = (coderClass) coderClass.getDeclaredConstructor().newInstance();
*/
List<String> parts = new ArrayList<>();
parts.add("");
parts.add(" = (");
parts.add(") ");
- parts.add(".newInstance();");
+ parts.add(".getDeclaredConstructor().newInstance();");
StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
List<Object> args = new ArrayList<>();
args.add(v1);
[beam] 05/24: Fix scala Product in Encoders to avoid StackEverflow
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a5d49f5dea33a862cc89ca8ef62c9a389a2902fd
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Aug 29 10:58:32 2019 +0200
Fix scala Product in Encoders to avoid StackEverflow
---
.../translation/helpers/EncoderHelpers.java | 18 ++++--------------
1 file changed, 4 insertions(+), 14 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 9cb8f29..8a4f1de 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -175,16 +175,11 @@ public class EncoderHelpers {
}
@Override public Object productElement(int n) {
- if (n == 0) {
- return this;
- } else {
- throw new IndexOutOfBoundsException(String.valueOf(n));
- }
+ return null;
}
@Override public int productArity() {
- //TODO test with spark Encoders if the arity of 1 is ok
- return 1;
+ return 0;
}
@Override public boolean canEqual(Object that) {
@@ -291,16 +286,11 @@ public class EncoderHelpers {
}
@Override public Object productElement(int n) {
- if (n == 0) {
- return this;
- } else {
- throw new IndexOutOfBoundsException(String.valueOf(n));
- }
+ return null;
}
@Override public int productArity() {
- //TODO test with spark Encoders if the arity of 1 is ok
- return 1;
+ return 0;
}
@Override public boolean canEqual(Object that) {
[beam] 02/24: Wrap Beam Coders into Spark Encoders using
ExpressionEncoder: deserialization part
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit d613d6bc28d63df6631367270f8474caf6f24059
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon Aug 26 15:22:12 2019 +0200
Wrap Beam Coders into Spark Encoders using ExpressionEncoder: deserialization part
---
.../translation/helpers/EncoderHelpers.java | 61 ++++++++++++++++++----
1 file changed, 50 insertions(+), 11 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index b072803..3159de9b 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -19,6 +19,8 @@ package org.apache.beam.runners.spark.structuredstreaming.translation.helpers;
import static org.apache.spark.sql.types.DataTypes.BinaryType;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;
@@ -103,7 +105,7 @@ public class EncoderHelpers {
SchemaHelpers.binarySchema(),
false,
JavaConversions.collectionAsScalaIterable(serialiserList).toSeq(),
- new DecodeUsingBeamCoder<>(classTag, coder), classTag);
+ new DecodeUsingBeamCoder<>(claz, coder), classTag);
/*
ExpressionEncoder[T](
@@ -150,8 +152,8 @@ public class EncoderHelpers {
List<String> instructions = new ArrayList<>();
instructions.add(outside);
-
Seq<String> parts = JavaConversions.collectionAsScalaIterable(instructions).toSeq();
+
StringContext stringContext = new StringContext(parts);
Block.BlockHelper blockHelper = new Block.BlockHelper(stringContext);
List<Object> args = new ArrayList<>();
@@ -160,7 +162,7 @@ public class EncoderHelpers {
args.add(new VariableValue("javaType", String.class));
args.add(new SimpleExprValue("input.isNull", Boolean.class));
args.add(new SimpleExprValue("CodeGenerator.defaultValue(dataType)", String.class));
- args.add(new VariableValue("$serialize", String.class));
+ args.add(new VariableValue("serialize", String.class));
Block code = blockHelper.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
return ev.copy(input.code().$plus(code), input.isNull(), new VariableValue("output", Array.class));
@@ -229,24 +231,61 @@ public class EncoderHelpers {
private static class DecodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression{
- private ClassTag<T> classTag;
+ private Class<T> claz;
private Coder<T> beamCoder;
+ private Expression child;
- private DecodeUsingBeamCoder(ClassTag<T> classTag, Coder<T> beamCoder) {
- this.classTag = classTag;
+ private DecodeUsingBeamCoder(Class<T> claz, Coder<T> beamCoder) {
+ this.claz = claz;
this.beamCoder = beamCoder;
+ this.child = new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType);
}
@Override public Expression child() {
- return new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType);
+ return child;
}
@Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
- return null;
+ // Code to deserialize.
+ ExprCode input = child.genCode(ctx);
+ String javaType = CodeGenerator.javaType(dataType());
+
+ String inputStream = "ByteArrayInputStream bais = new ByteArrayInputStream(${input.value});";
+ String deserialize = inputStream + "($javaType) $beamCoder.decode(bais);";
+
+ String outside = "final $javaType output = ${input.isNull} ? ${CodeGenerator.defaultValue(dataType)} : $deserialize;";
+
+ List<String> instructions = new ArrayList<>();
+ instructions.add(outside);
+ Seq<String> parts = JavaConversions.collectionAsScalaIterable(instructions).toSeq();
+
+ StringContext stringContext = new StringContext(parts);
+ Block.BlockHelper blockHelper = new Block.BlockHelper(stringContext);
+ List<Object> args = new ArrayList<>();
+ args.add(new SimpleExprValue("input.value", ExprValue.class));
+ args.add(new VariableValue("javaType", String.class));
+ args.add(new VariableValue("beamCoder", Coder.class));
+ args.add(new SimpleExprValue("input.isNull", Boolean.class));
+ args.add(new SimpleExprValue("CodeGenerator.defaultValue(dataType)", String.class));
+ args.add(new VariableValue("deserialize", String.class));
+ Block code = blockHelper.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+
+ return ev.copy(input.code().$plus(code), input.isNull(), new VariableValue("output", claz));
+
+ }
+
+ @Override public Object nullSafeEval(Object input) {
+ try {
+ return beamCoder.decode(new ByteArrayInputStream((byte[]) input));
+ } catch (IOException e) {
+ throw new IllegalStateException("Error decoding bytes for coder: " + beamCoder, e);
+ }
}
@Override public DataType dataType() {
- return new ObjectType(classTag.runtimeClass());
+// return new ObjectType(classTag.runtimeClass());
+ //TODO does type erasure impose to use classTag.runtimeClass() ?
+ return new ObjectType(claz);
}
@Override public Object productElement(int n) {
@@ -274,11 +313,11 @@ public class EncoderHelpers {
return false;
}
DecodeUsingBeamCoder<?> that = (DecodeUsingBeamCoder<?>) o;
- return classTag.equals(that.classTag) && beamCoder.equals(that.beamCoder);
+ return claz.equals(that.claz) && beamCoder.equals(that.beamCoder);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), classTag, beamCoder);
+ return Objects.hash(super.hashCode(), claz, beamCoder);
}
}
/*
[beam] 17/24: Cast coder instanciated by reflection
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 91e923c0dde332ea8615aa320b1c48d8c1167fc3
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 14:02:17 2019 +0200
Cast coder instanciated by reflection
---
.../structuredstreaming/translation/helpers/EncoderHelpers.java | 9 ++++++---
1 file changed, 6 insertions(+), 3 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 05595f1..0751c4c 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Objects;
import org.apache.beam.runners.spark.structuredstreaming.translation.SchemaHelpers;
import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.spark.sql.Encoder;
@@ -396,12 +397,14 @@ case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: B
*/
List<String> parts = new ArrayList<>();
parts.add("try {");
- parts.add(" = ");
- parts.add(".class.getDeclaredConstructor().newInstance();} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
+ parts.add(" = (");
+ parts.add(") ");
+ parts.add(".class.getDeclaredConstructor().newInstance();} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
List<Object> args = new ArrayList<>();
args.add(v1);
- args.add(coderClass.getName());
+ args.add(coderClass.getName());
+ args.add(coderClass.getName());
return sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
}));
return beamCoderInstance;
[beam] 07/24: Fix visibility of serializer and deserializer
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit e96af88096a6798ff0de2f24a95273334b3d9c7d
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Aug 29 15:18:44 2019 +0200
Fix visibility of serializer and deserializer
---
.../structuredstreaming/translation/helpers/EncoderHelpers.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 002545e..0765c78 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -126,12 +126,12 @@ public class EncoderHelpers {
*/
}
- private static class EncodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
+ public static class EncodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
private Expression child;
private Coder<T> beamCoder;
- private EncodeUsingBeamCoder(Expression child, Coder<T> beamCoder) {
+ public EncodeUsingBeamCoder(Expression child, Coder<T> beamCoder) {
this.child = child;
this.beamCoder = beamCoder;
}
@@ -231,13 +231,13 @@ public class EncoderHelpers {
override def dataType: DataType = BinaryType
}*/
- private static class DecodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression{
+ public static class DecodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression{
private Expression child;
private ClassTag<T> classTag;
private Coder<T> beamCoder;
- private DecodeUsingBeamCoder(Expression child, ClassTag<T> classTag, Coder<T> beamCoder) {
+ public DecodeUsingBeamCoder(Expression child, ClassTag<T> classTag, Coder<T> beamCoder) {
this.child = child;
this.classTag = classTag;
this.beamCoder = beamCoder;
[beam] 13/24: Fix ExpressionEncoder generated code: typos, try catch,
fqcn
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 668227b5240e08ff0de45ad7deeff1dad0a6a97d
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed Sep 4 15:38:41 2019 +0200
Fix ExpressionEncoder generated code: typos, try catch, fqcn
---
.../translation/helpers/EncoderHelpers.java | 38 +++++++++++++---------
1 file changed, 23 insertions(+), 15 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 1d89101..dff308a 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -144,18 +144,22 @@ public class EncoderHelpers {
/*
CODE GENERATED
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- final bytes[] output;
- if ({input.isNull})
- output = null;
- else
- output = $beamCoder.encode(${input.value}, baos); baos.toByteArray();
+ try {
+ java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream();
+ final byte[] output;
+ if ({input.isNull})
+ output = null;
+ else
+ output = $beamCoder.encode(${input.value}, baos); baos.toByteArray();
+ } catch (Exception e) {
+ throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
+ }
*/
List<String> parts = new ArrayList<>();
- parts.add("ByteArrayOutputStream baos = new ByteArrayOutputStream(); final bytes[] output; if (");
+ parts.add("try {java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); final byte[] output; if (");
parts.add(") output = null; else output =");
parts.add(".encode(");
- parts.add(", baos); baos.toByteArray();");
+ parts.add(", baos); baos.toByteArray();} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
@@ -258,21 +262,25 @@ public class EncoderHelpers {
/*
CODE GENERATED:
- final $javaType output =
- ${input.isNull} ?
- ${CodeGenerator.defaultValue(dataType)} :
- ($javaType) $beamCoder.decode(new ByteArrayInputStream(${input.value}));
+ try {
+ final $javaType output =
+ ${input.isNull} ?
+ ${CodeGenerator.defaultValue(dataType)} :
+ ($javaType) $beamCoder.decode(new java.io.ByteArrayInputStream(${input.value}));
+ } catch (IOException e) {
+ throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
+ }
*/
List<String> parts = new ArrayList<>();
- parts.add("final ");
+ parts.add("try { final ");
parts.add(" output =");
parts.add("?");
parts.add(":");
parts.add("(");
parts.add(") ");
- parts.add(".decode(new ByteArrayInputStream(");
- parts.add("));");
+ parts.add(".decode(new java.io.ByteArrayInputStream(");
+ parts.add(")); } catch (IOException e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
[beam] 19/24: Remove example code
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 43513041ded40c7d1ae0f6cf1adc81975aaa92c3
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 14:33:23 2019 +0200
Remove example code
---
.../translation/helpers/EncoderHelpers.java | 69 ----------------------
1 file changed, 69 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 3f7c102..83243b3 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -102,22 +102,6 @@ public class EncoderHelpers {
JavaConversions.collectionAsScalaIterable(serialiserList).toSeq(),
new DecodeUsingBeamCoder<>(new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, beamCoder),
classTag);
-
-/*
- ExpressionEncoder[T](
- schema = new StructType().add("value", BinaryType),
- flat = true,
- serializer = Seq(
- EncodeUsingSerializer(
- BoundReference(0, ObjectType(classOf[AnyRef]), nullable = true), kryo = useKryo)),
- deserializer =
- DecodeUsingSerializer[T](
- Cast(GetColumnByOrdinal(0, BinaryType), BinaryType),
- classTag[T],
- kryo = useKryo),
- clsTag = classTag[T]
- )
-*/
}
public static class EncodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
@@ -219,30 +203,6 @@ public class EncoderHelpers {
}
}
- /*case class EncodeUsingSerializer(child: Expression, kryo: Boolean)
- extends UnaryExpression with NonSQLExpression with SerializerSupport {
-
- override def nullSafeEval(input: Any): Any = {
- serializerInstance.serialize(input).array()
- }
-
- override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
- val serializer = addImmutableScodererializerIfNeeded(ctx)
- // Code to serialize.
- val input = child.genCode(ctx)
- val javaType = CodeGenerator.javaType(dataType)
- val serialize = s"$serializer.serialize(${input.value}, null).array()"
-
- val code = input.code + code"""
- final $javaType ${ev.value} =
- ${input.isNull} ? ${CodeGenerator.defaultValue(dataType)} : $serialize;
- """
- ev.copy(code = code, isNull = input.isNull)
- }
-
- override def dataType: DataType = BinaryType
- }*/
-
public static class DecodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression{
private Expression child;
@@ -353,33 +313,4 @@ public class EncoderHelpers {
return Objects.hash(super.hashCode(), classTag, beamCoder);
}
}
-/*
-case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: Boolean)
- extends UnaryExpression with NonSQLExpression with SerializerSupport {
-
- override def nullSafeEval(input: Any): Any = {
- val inputBytes = java.nio.ByteBuffer.wrap(input.asInstanceOf[Array[Byte]])
- serializerInstance.deserialize(inputBytes)
- }
-
- override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
- val serializer = addImmutableSerializerIfNeeded(ctx)
- // Code to deserialize.
- val input = child.genCode(ctx)
- val javaType = CodeGenerator.javaType(dataType)
- val deserialize =
- s"($javaType) $serializer.deserialize(java.nio.ByteBuffer.wrap(${input.value}), null)"
-
- val code = input.code + code"""
- final $javaType ${ev.value} =
- ${input.isNull} ? ${CodeGenerator.defaultValue(dataType)} : $deserialize;
- """
- ev.copy(code = code, isNull = input.isNull)
- }
-
- override def dataType: DataType = ObjectType(tag.runtimeClass)
- }
-*/
-
-
}
[beam] 09/24: Fix code generation in Beam coder wrapper
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit b11e1000c1dc122c2da315f7d6debf344a1d2055
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon Sep 2 15:45:24 2019 +0200
Fix code generation in Beam coder wrapper
---
.../translation/helpers/EncoderHelpers.java | 93 ++++++++++++----------
.../structuredstreaming/utils/EncodersTest.java | 4 +-
2 files changed, 55 insertions(+), 42 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 0765c78..cc862cd 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -42,15 +42,13 @@ import org.apache.spark.sql.catalyst.expressions.codegen.Block;
import org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator;
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext;
import org.apache.spark.sql.catalyst.expressions.codegen.ExprCode;
-import org.apache.spark.sql.catalyst.expressions.codegen.ExprValue;
-import org.apache.spark.sql.catalyst.expressions.codegen.SimpleExprValue;
import org.apache.spark.sql.catalyst.expressions.codegen.VariableValue;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.ObjectType;
+import scala.Function1;
import scala.StringContext;
import scala.Tuple2;
import scala.collection.JavaConversions;
-import scala.collection.Seq;
import scala.reflect.ClassTag;
import scala.reflect.ClassTag$;
@@ -143,29 +141,33 @@ public class EncoderHelpers {
@Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
// Code to serialize.
ExprCode input = child.genCode(ctx);
- String javaType = CodeGenerator.javaType(dataType());
- String outputStream = "ByteArrayOutputStream baos = new ByteArrayOutputStream();";
-
- String serialize = outputStream + "$beamCoder.encode(${input.value}, baos); baos.toByteArray();";
-
- String outside = "final $javaType output = ${input.isNull} ? ${CodeGenerator.defaultValue(dataType)} : $serialize;";
- List<String> instructions = new ArrayList<>();
- instructions.add(outside);
- Seq<String> parts = JavaConversions.collectionAsScalaIterable(instructions).toSeq();
+ /*
+ CODE GENERATED
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ final bytes[] output;
+ if ({input.isNull})
+ output = null;
+ else
+ output = $beamCoder.encode(${input.value}, baos); baos.toByteArray();
+ */
+ List<String> parts = new ArrayList<>();
+ parts.add("ByteArrayOutputStream baos = new ByteArrayOutputStream(); final bytes[] output; if (");
+ parts.add(") output = null; else output =");
+ parts.add(".encode(");
+ parts.add(", baos); baos.toByteArray();");
+
+ StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
- StringContext stringContext = new StringContext(parts);
- Block.BlockHelper blockHelper = new Block.BlockHelper(stringContext);
List<Object> args = new ArrayList<>();
- args.add(new VariableValue("beamCoder", Coder.class));
- args.add(new SimpleExprValue("input.value", ExprValue.class));
- args.add(new VariableValue("javaType", String.class));
- args.add(new SimpleExprValue("input.isNull", Boolean.class));
- args.add(new SimpleExprValue("CodeGenerator.defaultValue(dataType)", String.class));
- args.add(new VariableValue("serialize", String.class));
- Block code = blockHelper.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
-
- return ev.copy(input.code().$plus(code), input.isNull(), new VariableValue("output", Array.class));
+ args.add(input.isNull());
+ args.add(beamCoder);
+ args.add(input.value());
+ Block code = (new Block.BlockHelper(sc))
+ .code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+
+ return ev.copy(input.code().$plus(code), input.isNull(),
+ new VariableValue("output", Array.class));
}
@Override public DataType dataType() {
@@ -252,27 +254,38 @@ public class EncoderHelpers {
ExprCode input = child.genCode(ctx);
String javaType = CodeGenerator.javaType(dataType());
- String inputStream = "ByteArrayInputStream bais = new ByteArrayInputStream(${input.value});";
- String deserialize = inputStream + "($javaType) $beamCoder.decode(bais);";
+/*
+ CODE GENERATED:
+ final $javaType output =
+ ${input.isNull} ?
+ ${CodeGenerator.defaultValue(dataType)} :
+ ($javaType) $beamCoder.decode(new ByteArrayInputStream(${input.value}));
+*/
- String outside = "final $javaType output = ${input.isNull} ? ${CodeGenerator.defaultValue(dataType)} : $deserialize;";
+ List<String> parts = new ArrayList<>();
+ parts.add("final ");
+ parts.add(" output =");
+ parts.add("?");
+ parts.add(":");
+ parts.add("(");
+ parts.add(") ");
+ parts.add(".decode(new ByteArrayInputStream(");
+ parts.add("));");
- List<String> instructions = new ArrayList<>();
- instructions.add(outside);
- Seq<String> parts = JavaConversions.collectionAsScalaIterable(instructions).toSeq();
+ StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
- StringContext stringContext = new StringContext(parts);
- Block.BlockHelper blockHelper = new Block.BlockHelper(stringContext);
List<Object> args = new ArrayList<>();
- args.add(new SimpleExprValue("input.value", ExprValue.class));
- args.add(new VariableValue("javaType", String.class));
- args.add(new VariableValue("beamCoder", Coder.class));
- args.add(new SimpleExprValue("input.isNull", Boolean.class));
- args.add(new SimpleExprValue("CodeGenerator.defaultValue(dataType)", String.class));
- args.add(new VariableValue("deserialize", String.class));
- Block code = blockHelper.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
-
- return ev.copy(input.code().$plus(code), input.isNull(), new VariableValue("output", classTag.runtimeClass()));
+ args.add(javaType);
+ args.add(input.isNull());
+ args.add(CodeGenerator.defaultValue(dataType(), false));
+ args.add(javaType);
+ args.add(beamCoder);
+ args.add(input.value());
+ Block code = (new Block.BlockHelper(sc))
+ .code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+
+ return ev.copy(input.code().$plus(code), input.isNull(),
+ new VariableValue("output", classTag.runtimeClass()));
}
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
index 490e3dc..7078b0c 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
@@ -23,7 +23,7 @@ public class EncodersTest {
data.add(1);
data.add(2);
data.add(3);
-// sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
- sparkSession.createDataset(data, EncoderHelpers.genericEncoder());
+ sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
+// sparkSession.createDataset(data, EncoderHelpers.genericEncoder());
}
}
[beam] 22/24: Add an assert of equality in the encoders test
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 0fe6f9be8e7376f4481e09a13e2fdd0bed121a61
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 15:28:05 2019 +0200
Add an assert of equality in the encoders test
---
.../runners/spark/structuredstreaming/utils/EncodersTest.java | 10 +++++++---
1 file changed, 7 insertions(+), 3 deletions(-)
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
index b3a6273..c6b8631 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
@@ -1,10 +1,12 @@
package org.apache.beam.runners.spark.structuredstreaming.utils;
+import static org.junit.Assert.assertEquals;
+
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.spark.structuredstreaming.translation.helpers.EncoderHelpers;
-import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SparkSession;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -24,7 +26,9 @@ public class EncodersTest {
data.add(1);
data.add(2);
data.add(3);
- sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
-// sparkSession.createDataset(data, EncoderHelpers.genericEncoder());
+ Dataset<Integer> dataset = sparkSession
+ .createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
+ List<Integer> results = dataset.collectAsList();
+ assertEquals(data, results);
}
}
[beam] 01/24: Wrap Beam Coders into Spark Encoders using
ExpressionEncoder: serialization part
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 031754c73bcd72302f10730c8266e5ead0714bc1
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon Aug 26 14:32:17 2019 +0200
Wrap Beam Coders into Spark Encoders using ExpressionEncoder: serialization part
---
.../translation/helpers/EncoderHelpers.java | 245 +++++++++++++++++++++
1 file changed, 245 insertions(+)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index d44fe27..b072803 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -17,11 +17,40 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.helpers;
+import static org.apache.spark.sql.types.DataTypes.BinaryType;
+
+import java.lang.reflect.Array;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import org.apache.beam.runners.spark.structuredstreaming.translation.SchemaHelpers;
+import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.catalyst.analysis.GetColumnByOrdinal;
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder;
+import org.apache.spark.sql.catalyst.expressions.BoundReference;
+import org.apache.spark.sql.catalyst.expressions.Cast;
+import org.apache.spark.sql.catalyst.expressions.Expression;
+import org.apache.spark.sql.catalyst.expressions.NonSQLExpression;
+import org.apache.spark.sql.catalyst.expressions.UnaryExpression;
+import org.apache.spark.sql.catalyst.expressions.codegen.Block;
+import org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator;
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext;
+import org.apache.spark.sql.catalyst.expressions.codegen.ExprCode;
+import org.apache.spark.sql.catalyst.expressions.codegen.ExprValue;
+import org.apache.spark.sql.catalyst.expressions.codegen.SimpleExprValue;
+import org.apache.spark.sql.catalyst.expressions.codegen.VariableValue;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.ObjectType;
+import scala.StringContext;
import scala.Tuple2;
+import scala.collection.JavaConversions;
+import scala.collection.Seq;
+import scala.reflect.ClassTag;
+import scala.reflect.ClassTag$;
/** {@link Encoders} utility class. */
public class EncoderHelpers {
@@ -64,4 +93,220 @@ public class EncoderHelpers {
--------- Bridges from Beam Coders to Spark Encoders
*/
+ /** A way to construct encoders using generic serializers. */
+ private <T> Encoder<T> fromBeamCoder(Coder<T> coder, Class<T> claz){
+
+ List<Expression> serialiserList = new ArrayList<>();
+ serialiserList.add(new EncodeUsingBeamCoder<>(claz, coder));
+ ClassTag<T> classTag = ClassTag$.MODULE$.apply(claz);
+ return new ExpressionEncoder<>(
+ SchemaHelpers.binarySchema(),
+ false,
+ JavaConversions.collectionAsScalaIterable(serialiserList).toSeq(),
+ new DecodeUsingBeamCoder<>(classTag, coder), classTag);
+
+/*
+ ExpressionEncoder[T](
+ schema = new StructType().add("value", BinaryType),
+ flat = true,
+ serializer = Seq(
+ EncodeUsingSerializer(
+ BoundReference(0, ObjectType(classOf[AnyRef]), nullable = true), kryo = useKryo)),
+ deserializer =
+ DecodeUsingSerializer[T](
+ Cast(GetColumnByOrdinal(0, BinaryType), BinaryType),
+ classTag[T],
+ kryo = useKryo),
+ clsTag = classTag[T]
+ )
+*/
+ }
+
+ private static class EncodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
+
+ private Class<T> claz;
+ private Coder<T> beamCoder;
+ private Expression child;
+
+ private EncodeUsingBeamCoder( Class<T> claz, Coder<T> beamCoder) {
+ this.claz = claz;
+ this.beamCoder = beamCoder;
+ this.child = new BoundReference(0, new ObjectType(claz), true);
+ }
+
+ @Override public Expression child() {
+ return child;
+ }
+
+ @Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
+ // Code to serialize.
+ ExprCode input = child.genCode(ctx);
+ String javaType = CodeGenerator.javaType(dataType());
+ String outputStream = "ByteArrayOutputStream baos = new ByteArrayOutputStream();";
+
+ String serialize = outputStream + "$beamCoder.encode(${input.value}, baos); baos.toByteArray();";
+
+ String outside = "final $javaType output = ${input.isNull} ? ${CodeGenerator.defaultValue(dataType)} : $serialize;";
+
+ List<String> instructions = new ArrayList<>();
+ instructions.add(outside);
+
+ Seq<String> parts = JavaConversions.collectionAsScalaIterable(instructions).toSeq();
+ StringContext stringContext = new StringContext(parts);
+ Block.BlockHelper blockHelper = new Block.BlockHelper(stringContext);
+ List<Object> args = new ArrayList<>();
+ args.add(new VariableValue("beamCoder", Coder.class));
+ args.add(new SimpleExprValue("input.value", ExprValue.class));
+ args.add(new VariableValue("javaType", String.class));
+ args.add(new SimpleExprValue("input.isNull", Boolean.class));
+ args.add(new SimpleExprValue("CodeGenerator.defaultValue(dataType)", String.class));
+ args.add(new VariableValue("$serialize", String.class));
+ Block code = blockHelper.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+
+ return ev.copy(input.code().$plus(code), input.isNull(), new VariableValue("output", Array.class));
+ }
+
+ @Override public DataType dataType() {
+ return BinaryType;
+ }
+
+ @Override public Object productElement(int n) {
+ if (n == 0) {
+ return this;
+ } else {
+ throw new IndexOutOfBoundsException(String.valueOf(n));
+ }
+ }
+
+ @Override public int productArity() {
+ //TODO test with spark Encoders if the arity of 1 is ok
+ return 1;
+ }
+
+ @Override public boolean canEqual(Object that) {
+ return (that instanceof EncodeUsingBeamCoder);
+ }
+
+ @Override public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ EncodeUsingBeamCoder<?> that = (EncodeUsingBeamCoder<?>) o;
+ return claz.equals(that.claz) && beamCoder.equals(that.beamCoder);
+ }
+
+ @Override public int hashCode() {
+ return Objects.hash(super.hashCode(), claz, beamCoder);
+ }
+ }
+
+ /*case class EncodeUsingSerializer(child: Expression, kryo: Boolean)
+ extends UnaryExpression with NonSQLExpression with SerializerSupport {
+
+ override def nullSafeEval(input: Any): Any = {
+ serializerInstance.serialize(input).array()
+ }
+
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+ val serializer = addImmutableScodererializerIfNeeded(ctx)
+ // Code to serialize.
+ val input = child.genCode(ctx)
+ val javaType = CodeGenerator.javaType(dataType)
+ val serialize = s"$serializer.serialize(${input.value}, null).array()"
+
+ val code = input.code + code"""
+ final $javaType ${ev.value} =
+ ${input.isNull} ? ${CodeGenerator.defaultValue(dataType)} : $serialize;
+ """
+ ev.copy(code = code, isNull = input.isNull)
+ }
+
+ override def dataType: DataType = BinaryType
+ }*/
+
+ private static class DecodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression{
+
+ private ClassTag<T> classTag;
+ private Coder<T> beamCoder;
+
+ private DecodeUsingBeamCoder(ClassTag<T> classTag, Coder<T> beamCoder) {
+ this.classTag = classTag;
+ this.beamCoder = beamCoder;
+ }
+
+ @Override public Expression child() {
+ return new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType);
+ }
+
+ @Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
+ return null;
+ }
+
+ @Override public DataType dataType() {
+ return new ObjectType(classTag.runtimeClass());
+ }
+
+ @Override public Object productElement(int n) {
+ if (n == 0) {
+ return this;
+ } else {
+ throw new IndexOutOfBoundsException(String.valueOf(n));
+ }
+ }
+
+ @Override public int productArity() {
+ //TODO test with spark Encoders if the arity of 1 is ok
+ return 1;
+ }
+
+ @Override public boolean canEqual(Object that) {
+ return (that instanceof DecodeUsingBeamCoder);
+ }
+
+ @Override public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DecodeUsingBeamCoder<?> that = (DecodeUsingBeamCoder<?>) o;
+ return classTag.equals(that.classTag) && beamCoder.equals(that.beamCoder);
+ }
+
+ @Override public int hashCode() {
+ return Objects.hash(super.hashCode(), classTag, beamCoder);
+ }
+ }
+/*
+case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: Boolean)
+ extends UnaryExpression with NonSQLExpression with SerializerSupport {
+
+ override def nullSafeEval(input: Any): Any = {
+ val inputBytes = java.nio.ByteBuffer.wrap(input.asInstanceOf[Array[Byte]])
+ serializerInstance.deserialize(inputBytes)
+ }
+
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+ val serializer = addImmutableSerializerIfNeeded(ctx)
+ // Code to deserialize.
+ val input = child.genCode(ctx)
+ val javaType = CodeGenerator.javaType(dataType)
+ val deserialize =
+ s"($javaType) $serializer.deserialize(java.nio.ByteBuffer.wrap(${input.value}), null)"
+
+ val code = input.code + code"""
+ final $javaType ${ev.value} =
+ ${input.isNull} ? ${CodeGenerator.defaultValue(dataType)} : $deserialize;
+ """
+ ev.copy(code = code, isNull = input.isNull)
+ }
+
+ override def dataType: DataType = ObjectType(tag.runtimeClass)
+ }
+*/
+
}
[beam] 11/24: Fix call to scala Fucntion1 in coder lazy init
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 2c94eef2f1d2677bf64851a8c22520c44557170f
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed Sep 4 14:54:59 2019 +0200
Fix call to scala Fucntion1 in coder lazy init
---
runners/spark/build.gradle | 1 +
.../translation/helpers/EncoderHelpers.java | 28 +++++++++++-----------
2 files changed, 15 insertions(+), 14 deletions(-)
diff --git a/runners/spark/build.gradle b/runners/spark/build.gradle
index 73a710b..a948ef1 100644
--- a/runners/spark/build.gradle
+++ b/runners/spark/build.gradle
@@ -77,6 +77,7 @@ dependencies {
provided "com.esotericsoftware.kryo:kryo:2.21"
runtimeOnly library.java.jackson_module_scala
runtimeOnly "org.scala-lang:scala-library:2.11.8"
+ compile "org.scala-lang.modules:scala-java8-compat_2.11:0.9.0"
testCompile project(":sdks:java:io:kafka")
testCompile project(path: ":sdks:java:core", configuration: "shadowTest")
// SparkStateInternalsTest extends abstract StateInternalsTest
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index f7706cc..694bc24 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -18,9 +18,9 @@
package org.apache.beam.runners.spark.structuredstreaming.translation.helpers;
import static org.apache.spark.sql.types.DataTypes.BinaryType;
+import static scala.compat.java8.JFunction.func;
import java.io.ByteArrayInputStream;
-import java.io.IOException;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;
@@ -370,23 +370,23 @@ case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: B
private static <T> String lazyInitBeamCoder(CodegenContext ctx, Class<Coder<T>> coderClass) {
String beamCoderInstance = "beamCoder";
- ctx.addImmutableStateIfNotExists(coderClass.getName(), beamCoderInstance, v -> {
+ ctx.addImmutableStateIfNotExists(coderClass.getName(), beamCoderInstance, func(v1 -> {
/*
CODE GENERATED
v = (coderClass) coderClass.newInstance();
*/
- List<String> parts = new ArrayList<>();
- parts.add("");
- parts.add(" = (");
- parts.add(") ");
- parts.add(".newInstance();");
- StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
- List<Object> args = new ArrayList<>();
- args.add(v);
- args.add(coderClass.getName());
- args.add(coderClass.getName());
- return sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
- });
+ List<String> parts = new ArrayList<>();
+ parts.add("");
+ parts.add(" = (");
+ parts.add(") ");
+ parts.add(".newInstance();");
+ StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
+ List<Object> args = new ArrayList<>();
+ args.add(v1);
+ args.add(coderClass.getName());
+ args.add(coderClass.getName());
+ return sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
+ }));
return beamCoderInstance;
}
[beam] 16/24: Add try catch around reflexion call in lazy init of
beam coder
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 723c004dcdf82073568dc1a758d8d7453a6af954
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 10:13:05 2019 +0200
Add try catch around reflexion call in lazy init of beam coder
---
.../structuredstreaming/translation/helpers/EncoderHelpers.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 56097b7..05595f1 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -388,12 +388,16 @@ case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: B
ctx.addImmutableStateIfNotExists(coderClass.getName(), beamCoderInstance, func(v1 -> {
/*
CODE GENERATED
+ try {
v1 = coderClass.class.getDeclaredConstructor().newInstance();
+ } catch (Exception e) {
+ throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
+ }
*/
List<String> parts = new ArrayList<>();
- parts.add("");
+ parts.add("try {");
parts.add(" = ");
- parts.add(".class.getDeclaredConstructor().newInstance();");
+ parts.add(".class.getDeclaredConstructor().newInstance();} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
List<Object> args = new ArrayList<>();
args.add(v1);
[beam] 03/24: Fix EncoderHelpers.fromBeamCoder() visibility
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 84f2cbd9b30a72ecc827e9c03de533b8e65190b2
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue Aug 27 17:26:30 2019 +0200
Fix EncoderHelpers.fromBeamCoder() visibility
---
.../structuredstreaming/translation/helpers/EncoderHelpers.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 3159de9b..ab24e37 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -96,7 +96,7 @@ public class EncoderHelpers {
*/
/** A way to construct encoders using generic serializers. */
- private <T> Encoder<T> fromBeamCoder(Coder<T> coder, Class<T> claz){
+ public static <T> Encoder<T> fromBeamCoder(Coder<T> coder, Class<T> claz){
List<Expression> serialiserList = new ArrayList<>();
serialiserList.add(new EncodeUsingBeamCoder<>(claz, coder));
@@ -105,7 +105,8 @@ public class EncoderHelpers {
SchemaHelpers.binarySchema(),
false,
JavaConversions.collectionAsScalaIterable(serialiserList).toSeq(),
- new DecodeUsingBeamCoder<>(claz, coder), classTag);
+ new DecodeUsingBeamCoder<>(claz, coder),
+ classTag);
/*
ExpressionEncoder[T](
[beam] 06/24: Conform to spark ExpressionEncoders: pass classTags,
implement scala Product, pass children from within the ExpressionEncoder
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 23735e4bc4cc6fe7a0e56a5370499d0ce95d381d
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Aug 29 15:10:40 2019 +0200
Conform to spark ExpressionEncoders: pass classTags, implement scala Product, pass children from within the ExpressionEncoder
---
.../translation/helpers/EncoderHelpers.java | 60 +++++++++++++---------
1 file changed, 36 insertions(+), 24 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 8a4f1de..002545e 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -100,13 +100,13 @@ public class EncoderHelpers {
List<Expression> serialiserList = new ArrayList<>();
Class<T> claz = (Class<T>) Object.class;
- serialiserList.add(new EncodeUsingBeamCoder<>(claz, coder));
+ serialiserList.add(new EncodeUsingBeamCoder<>(new BoundReference(0, new ObjectType(claz), true), coder));
ClassTag<T> classTag = ClassTag$.MODULE$.apply(claz);
return new ExpressionEncoder<>(
SchemaHelpers.binarySchema(),
false,
JavaConversions.collectionAsScalaIterable(serialiserList).toSeq(),
- new DecodeUsingBeamCoder<>(claz, coder),
+ new DecodeUsingBeamCoder<>(new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, coder),
classTag);
/*
@@ -128,14 +128,12 @@ public class EncoderHelpers {
private static class EncodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
- private Class<T> claz;
- private Coder<T> beamCoder;
private Expression child;
+ private Coder<T> beamCoder;
- private EncodeUsingBeamCoder( Class<T> claz, Coder<T> beamCoder) {
- this.claz = claz;
+ private EncodeUsingBeamCoder(Expression child, Coder<T> beamCoder) {
+ this.child = child;
this.beamCoder = beamCoder;
- this.child = new BoundReference(0, new ObjectType(claz), true);
}
@Override public Expression child() {
@@ -175,11 +173,18 @@ public class EncoderHelpers {
}
@Override public Object productElement(int n) {
- return null;
+ switch (n) {
+ case 0:
+ return child;
+ case 1:
+ return beamCoder;
+ default:
+ throw new ArrayIndexOutOfBoundsException("productElement out of bounds");
+ }
}
@Override public int productArity() {
- return 0;
+ return 2;
}
@Override public boolean canEqual(Object that) {
@@ -194,11 +199,11 @@ public class EncoderHelpers {
return false;
}
EncodeUsingBeamCoder<?> that = (EncodeUsingBeamCoder<?>) o;
- return claz.equals(that.claz) && beamCoder.equals(that.beamCoder);
+ return beamCoder.equals(that.beamCoder);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), claz, beamCoder);
+ return Objects.hash(super.hashCode(), beamCoder);
}
}
@@ -228,14 +233,14 @@ public class EncoderHelpers {
private static class DecodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression{
- private Class<T> claz;
- private Coder<T> beamCoder;
private Expression child;
+ private ClassTag<T> classTag;
+ private Coder<T> beamCoder;
- private DecodeUsingBeamCoder(Class<T> claz, Coder<T> beamCoder) {
- this.claz = claz;
+ private DecodeUsingBeamCoder(Expression child, ClassTag<T> classTag, Coder<T> beamCoder) {
+ this.child = child;
+ this.classTag = classTag;
this.beamCoder = beamCoder;
- this.child = new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType);
}
@Override public Expression child() {
@@ -267,7 +272,7 @@ public class EncoderHelpers {
args.add(new VariableValue("deserialize", String.class));
Block code = blockHelper.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
- return ev.copy(input.code().$plus(code), input.isNull(), new VariableValue("output", claz));
+ return ev.copy(input.code().$plus(code), input.isNull(), new VariableValue("output", classTag.runtimeClass()));
}
@@ -280,17 +285,24 @@ public class EncoderHelpers {
}
@Override public DataType dataType() {
-// return new ObjectType(classTag.runtimeClass());
- //TODO does type erasure impose to use classTag.runtimeClass() ?
- return new ObjectType(claz);
+ return new ObjectType(classTag.runtimeClass());
}
@Override public Object productElement(int n) {
- return null;
+ switch (n) {
+ case 0:
+ return child;
+ case 1:
+ return classTag;
+ case 2:
+ return beamCoder;
+ default:
+ throw new ArrayIndexOutOfBoundsException("productElement out of bounds");
+ }
}
@Override public int productArity() {
- return 0;
+ return 3;
}
@Override public boolean canEqual(Object that) {
@@ -305,11 +317,11 @@ public class EncoderHelpers {
return false;
}
DecodeUsingBeamCoder<?> that = (DecodeUsingBeamCoder<?>) o;
- return claz.equals(that.claz) && beamCoder.equals(that.beamCoder);
+ return classTag.equals(that.classTag) && beamCoder.equals(that.beamCoder);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), claz, beamCoder);
+ return Objects.hash(super.hashCode(), classTag, beamCoder);
}
}
/*
[beam] 21/24: Fix generated code: uniform exceptions catching,
fix parenthesis and variable declarations
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit d8b8b428617860c65ddf856e73fd3418731f4b1f
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 15:14:32 2019 +0200
Fix generated code: uniform exceptions catching, fix parenthesis and variable declarations
---
.../translation/helpers/EncoderHelpers.java | 18 ++++++++++--------
1 file changed, 10 insertions(+), 8 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 91aaaf9..c9ab435 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -227,32 +227,34 @@ public class EncoderHelpers {
/*
CODE GENERATED:
+ final $javaType ${ev.value}
try {
- final $javaType ${ev.value} =
+ ${ev.value} =
${input.isNull} ?
${CodeGenerator.defaultValue(dataType)} :
($javaType) $beamCoder.decode(new java.io.ByteArrayInputStream(${input.value}));
- } catch (IOException e) {
+ } catch (Exception e) {
throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
}
*/
List<String> parts = new ArrayList<>();
- parts.add("try { final ");
+ parts.add("final ");
parts.add(" ");
- parts.add(" =");
- parts.add("?");
- parts.add(":");
- parts.add("(");
+ parts.add(";try { ");
+ parts.add(" = ");
+ parts.add("? ");
+ parts.add(": (");
parts.add(") ");
parts.add(".decode(new java.io.ByteArrayInputStream(");
- parts.add(")); } catch (IOException e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
+ parts.add(")); } catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
List<Object> args = new ArrayList<>();
args.add(javaType);
args.add(ev.value());
+ args.add(ev.value());
args.add(input.isNull());
args.add(CodeGenerator.defaultValue(dataType(), false));
args.add(javaType);
[beam] 04/24: type erasure: spark encoders require a Class,
pass Object and cast to Class
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 95fd2727f9e2ad00144e790a8dd324d60867d35c
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Aug 29 10:57:53 2019 +0200
type erasure: spark encoders require a Class<T>, pass Object and cast to Class<T>
---
.../spark/structuredstreaming/translation/helpers/EncoderHelpers.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index ab24e37..9cb8f29 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -96,9 +96,10 @@ public class EncoderHelpers {
*/
/** A way to construct encoders using generic serializers. */
- public static <T> Encoder<T> fromBeamCoder(Coder<T> coder, Class<T> claz){
+ public static <T> Encoder<T> fromBeamCoder(Coder<T> coder/*, Class<T> claz*/){
List<Expression> serialiserList = new ArrayList<>();
+ Class<T> claz = (Class<T>) Object.class;
serialiserList.add(new EncodeUsingBeamCoder<>(claz, coder));
ClassTag<T> classTag = ClassTag$.MODULE$.apply(claz);
return new ExpressionEncoder<>(
[beam] 15/24: Fix beam coder lazy init using reflexion
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 8bbf991e4bded15e447f8bee7ef9e70ce9ff3ad4
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 10:07:32 2019 +0200
Fix beam coder lazy init using reflexion
---
.../translation/helpers/EncoderHelpers.java | 10 ++++------
1 file changed, 4 insertions(+), 6 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index a452da0..56097b7 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -388,18 +388,16 @@ case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: B
ctx.addImmutableStateIfNotExists(coderClass.getName(), beamCoderInstance, func(v1 -> {
/*
CODE GENERATED
- v = (coderClass) coderClass.getDeclaredConstructor().newInstance();
+ v1 = coderClass.class.getDeclaredConstructor().newInstance();
*/
- List<String> parts = new ArrayList<>();
+ List<String> parts = new ArrayList<>();
parts.add("");
- parts.add(" = (");
- parts.add(") ");
- parts.add(".getDeclaredConstructor().newInstance();");
+ parts.add(" = ");
+ parts.add(".class.getDeclaredConstructor().newInstance();");
StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
List<Object> args = new ArrayList<>();
args.add(v1);
args.add(coderClass.getName());
- args.add(coderClass.getName());
return sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
}));
return beamCoderInstance;
[beam] 14/24: Fix getting the output value in code generation
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 959664f8092e5bb06c7a78dae78a12ac2559d413
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed Sep 4 16:50:17 2019 +0200
Fix getting the output value in code generation
---
.../translation/helpers/EncoderHelpers.java | 37 +++++++++++++---------
1 file changed, 22 insertions(+), 15 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index dff308a..a452da0 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -21,7 +21,6 @@ import static org.apache.spark.sql.types.DataTypes.BinaryType;
import static scala.compat.java8.JFunction.func;
import java.io.ByteArrayInputStream;
-import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
@@ -42,7 +41,6 @@ import org.apache.spark.sql.catalyst.expressions.codegen.Block;
import org.apache.spark.sql.catalyst.expressions.codegen.CodeGenerator;
import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext;
import org.apache.spark.sql.catalyst.expressions.codegen.ExprCode;
-import org.apache.spark.sql.catalyst.expressions.codegen.VariableValue;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.ObjectType;
import scala.StringContext;
@@ -144,34 +142,42 @@ public class EncoderHelpers {
/*
CODE GENERATED
+ byte[] ${ev.value};
try {
java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream();
- final byte[] output;
if ({input.isNull})
- output = null;
- else
- output = $beamCoder.encode(${input.value}, baos); baos.toByteArray();
+ ${ev.value} = null;
+ else{
+ $beamCoder.encode(${input.value}, baos);
+ ${ev.value} = baos.toByteArray();
+ }
} catch (Exception e) {
throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
}
*/
List<String> parts = new ArrayList<>();
- parts.add("try {java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); final byte[] output; if (");
- parts.add(") output = null; else output =");
+ parts.add("byte[] ");
+ parts.add(";try { java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); if (");
+ parts.add(") ");
+ parts.add(" = null; else{");
parts.add(".encode(");
- parts.add(", baos); baos.toByteArray();} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
+ parts.add(", baos); ");
+ parts.add(" = baos.toByteArray();}} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
List<Object> args = new ArrayList<>();
+
+ args.add(ev.value());
args.add(input.isNull());
+ args.add(ev.value());
args.add(beamCoder);
args.add(input.value());
+ args.add(ev.value());
Block code = (new Block.BlockHelper(sc))
.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
- return ev.copy(input.code().$plus(code), input.isNull(),
- new VariableValue("output", Array.class));
+ return ev.copy(input.code().$plus(code), input.isNull(),ev.value());
}
@@ -263,7 +269,7 @@ public class EncoderHelpers {
/*
CODE GENERATED:
try {
- final $javaType output =
+ final $javaType ${ev.value} =
${input.isNull} ?
${CodeGenerator.defaultValue(dataType)} :
($javaType) $beamCoder.decode(new java.io.ByteArrayInputStream(${input.value}));
@@ -274,7 +280,8 @@ public class EncoderHelpers {
List<String> parts = new ArrayList<>();
parts.add("try { final ");
- parts.add(" output =");
+ parts.add(" ");
+ parts.add(" =");
parts.add("?");
parts.add(":");
parts.add("(");
@@ -286,6 +293,7 @@ public class EncoderHelpers {
List<Object> args = new ArrayList<>();
args.add(javaType);
+ args.add(ev.value());
args.add(input.isNull());
args.add(CodeGenerator.defaultValue(dataType(), false));
args.add(javaType);
@@ -294,8 +302,7 @@ public class EncoderHelpers {
Block code = (new Block.BlockHelper(sc))
.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
- return ev.copy(input.code().$plus(code), input.isNull(),
- new VariableValue("output", classTag.runtimeClass()));
+ return ev.copy(input.code().$plus(code), input.isNull(), ev.value());
}
[beam] 23/24: Apply spotless and checkstyle and add javadocs
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit b87912300c863af606933a03ca790e8e94ed05f1
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 15:35:34 2019 +0200
Apply spotless and checkstyle and add javadocs
---
.../translation/helpers/EncoderHelpers.java | 137 +++++++++++++--------
.../structuredstreaming/utils/EncodersTest.java | 32 +++--
2 files changed, 113 insertions(+), 56 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index c9ab435..f990121 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -89,21 +89,31 @@ public class EncoderHelpers {
--------- Bridges from Beam Coders to Spark Encoders
*/
- /** A way to construct encoders using generic serializers. */
- public static <T> Encoder<T> fromBeamCoder(Coder<T> beamCoder/*, Class<T> claz*/){
+ /**
+ * Wrap a Beam coder into a Spark Encoder using Catalyst Expression Encoders (which uses java code
+ * generation).
+ */
+ public static <T> Encoder<T> fromBeamCoder(Coder<T> beamCoder) {
List<Expression> serialiserList = new ArrayList<>();
Class<T> claz = (Class<T>) Object.class;
- serialiserList.add(new EncodeUsingBeamCoder<>(new BoundReference(0, new ObjectType(claz), true), beamCoder));
+ serialiserList.add(
+ new EncodeUsingBeamCoder<>(new BoundReference(0, new ObjectType(claz), true), beamCoder));
ClassTag<T> classTag = ClassTag$.MODULE$.apply(claz);
return new ExpressionEncoder<>(
SchemaHelpers.binarySchema(),
false,
JavaConversions.collectionAsScalaIterable(serialiserList).toSeq(),
- new DecodeUsingBeamCoder<>(new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, beamCoder),
+ new DecodeUsingBeamCoder<>(
+ new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, beamCoder),
classTag);
}
+ /**
+ * Catalyst Expression that serializes elements using Beam {@link Coder}.
+ *
+ * @param <T>: Type of elements ot be serialized.
+ */
public static class EncodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
private Expression child;
@@ -114,13 +124,16 @@ public class EncoderHelpers {
this.beamCoder = beamCoder;
}
- @Override public Expression child() {
+ @Override
+ public Expression child() {
return child;
}
- @Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
+ @Override
+ public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
// Code to serialize.
- String accessCode = ctx.addReferenceObj("beamCoder", beamCoder, beamCoder.getClass().getName());
+ String accessCode =
+ ctx.addReferenceObj("beamCoder", beamCoder, beamCoder.getClass().getName());
ExprCode input = child.genCode(ctx);
/*
@@ -140,14 +153,17 @@ public class EncoderHelpers {
*/
List<String> parts = new ArrayList<>();
parts.add("byte[] ");
- parts.add(";try { java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); if (");
+ parts.add(
+ ";try { java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream(); if (");
parts.add(") ");
parts.add(" = null; else{");
parts.add(".encode(");
parts.add(", baos); ");
- parts.add(" = baos.toByteArray();}} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
+ parts.add(
+ " = baos.toByteArray();}} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
- StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
+ StringContext sc =
+ new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
List<Object> args = new ArrayList<>();
@@ -157,18 +173,19 @@ public class EncoderHelpers {
args.add(accessCode);
args.add(input.value());
args.add(ev.value());
- Block code = (new Block.BlockHelper(sc))
- .code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+ Block code =
+ (new Block.BlockHelper(sc)).code(JavaConversions.collectionAsScalaIterable(args).toSeq());
- return ev.copy(input.code().$plus(code), input.isNull(),ev.value());
+ return ev.copy(input.code().$plus(code), input.isNull(), ev.value());
}
-
- @Override public DataType dataType() {
+ @Override
+ public DataType dataType() {
return BinaryType;
}
- @Override public Object productElement(int n) {
+ @Override
+ public Object productElement(int n) {
switch (n) {
case 0:
return child;
@@ -179,15 +196,18 @@ public class EncoderHelpers {
}
}
- @Override public int productArity() {
+ @Override
+ public int productArity() {
return 2;
}
- @Override public boolean canEqual(Object that) {
+ @Override
+ public boolean canEqual(Object that) {
return (that instanceof EncodeUsingBeamCoder);
}
- @Override public boolean equals(Object o) {
+ @Override
+ public boolean equals(Object o) {
if (this == o) {
return true;
}
@@ -198,12 +218,18 @@ public class EncoderHelpers {
return beamCoder.equals(that.beamCoder) && child.equals(that.child);
}
- @Override public int hashCode() {
+ @Override
+ public int hashCode() {
return Objects.hash(super.hashCode(), child, beamCoder);
}
}
- public static class DecodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression{
+ /**
+ * Catalyst Expression that deserializes elements using Beam {@link Coder}.
+ *
+ * @param <T>: Type of elements ot be serialized.
+ */
+ public static class DecodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
private Expression child;
private ClassTag<T> classTag;
@@ -215,28 +241,31 @@ public class EncoderHelpers {
this.beamCoder = beamCoder;
}
- @Override public Expression child() {
+ @Override
+ public Expression child() {
return child;
}
- @Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
+ @Override
+ public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
// Code to deserialize.
- String accessCode = ctx.addReferenceObj("beamCoder", beamCoder, beamCoder.getClass().getName());
+ String accessCode =
+ ctx.addReferenceObj("beamCoder", beamCoder, beamCoder.getClass().getName());
ExprCode input = child.genCode(ctx);
String javaType = CodeGenerator.javaType(dataType());
-/*
- CODE GENERATED:
- final $javaType ${ev.value}
- try {
- ${ev.value} =
- ${input.isNull} ?
- ${CodeGenerator.defaultValue(dataType)} :
- ($javaType) $beamCoder.decode(new java.io.ByteArrayInputStream(${input.value}));
- } catch (Exception e) {
- throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
- }
-*/
+ /*
+ CODE GENERATED:
+ final $javaType ${ev.value}
+ try {
+ ${ev.value} =
+ ${input.isNull} ?
+ ${CodeGenerator.defaultValue(dataType)} :
+ ($javaType) $beamCoder.decode(new java.io.ByteArrayInputStream(${input.value}));
+ } catch (Exception e) {
+ throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
+ }
+ */
List<String> parts = new ArrayList<>();
parts.add("final ");
@@ -247,9 +276,11 @@ public class EncoderHelpers {
parts.add(": (");
parts.add(") ");
parts.add(".decode(new java.io.ByteArrayInputStream(");
- parts.add(")); } catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
+ parts.add(
+ ")); } catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
- StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
+ StringContext sc =
+ new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
List<Object> args = new ArrayList<>();
args.add(javaType);
@@ -260,14 +291,14 @@ public class EncoderHelpers {
args.add(javaType);
args.add(accessCode);
args.add(input.value());
- Block code = (new Block.BlockHelper(sc))
- .code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+ Block code =
+ (new Block.BlockHelper(sc)).code(JavaConversions.collectionAsScalaIterable(args).toSeq());
return ev.copy(input.code().$plus(code), input.isNull(), ev.value());
-
}
- @Override public Object nullSafeEval(Object input) {
+ @Override
+ public Object nullSafeEval(Object input) {
try {
return beamCoder.decode(new ByteArrayInputStream((byte[]) input));
} catch (Exception e) {
@@ -275,11 +306,13 @@ public class EncoderHelpers {
}
}
- @Override public DataType dataType() {
+ @Override
+ public DataType dataType() {
return new ObjectType(classTag.runtimeClass());
}
- @Override public Object productElement(int n) {
+ @Override
+ public Object productElement(int n) {
switch (n) {
case 0:
return child;
@@ -292,15 +325,18 @@ public class EncoderHelpers {
}
}
- @Override public int productArity() {
+ @Override
+ public int productArity() {
return 3;
}
- @Override public boolean canEqual(Object that) {
+ @Override
+ public boolean canEqual(Object that) {
return (that instanceof DecodeUsingBeamCoder);
}
- @Override public boolean equals(Object o) {
+ @Override
+ public boolean equals(Object o) {
if (this == o) {
return true;
}
@@ -308,10 +344,13 @@ public class EncoderHelpers {
return false;
}
DecodeUsingBeamCoder<?> that = (DecodeUsingBeamCoder<?>) o;
- return child.equals(that.child) && classTag.equals(that.classTag) && beamCoder.equals(that.beamCoder);
+ return child.equals(that.child)
+ && classTag.equals(that.classTag)
+ && beamCoder.equals(that.beamCoder);
}
- @Override public int hashCode() {
+ @Override
+ public int hashCode() {
return Objects.hash(super.hashCode(), child, classTag, beamCoder);
}
}
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
index c6b8631..8327fd8 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
@@ -1,3 +1,20 @@
+/*
+ * 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.runners.spark.structuredstreaming.utils;
import static org.junit.Assert.assertEquals;
@@ -12,22 +29,23 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
+/** Test of the wrapping of Beam Coders as Spark ExpressionEncoders. */
@RunWith(JUnit4.class)
-/**
- * Test of the wrapping of Beam Coders as Spark ExpressionEncoders.
- */
public class EncodersTest {
@Test
public void beamCoderToSparkEncoderTest() {
- SparkSession sparkSession = SparkSession.builder().appName("beamCoderToSparkEncoderTest")
- .master("local[4]").getOrCreate();
+ SparkSession sparkSession =
+ SparkSession.builder()
+ .appName("beamCoderToSparkEncoderTest")
+ .master("local[4]")
+ .getOrCreate();
List<Integer> data = new ArrayList<>();
data.add(1);
data.add(2);
data.add(3);
- Dataset<Integer> dataset = sparkSession
- .createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
+ Dataset<Integer> dataset =
+ sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
List<Integer> results = dataset.collectAsList();
assertEquals(data, results);
}
[beam] 18/24: Remove lazy init of beam coder because there is no
generic way on instanciating a beam coder
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit c4a44649013b04d04eeca3756ff181f59632fab7
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Thu Sep 5 14:20:30 2019 +0200
Remove lazy init of beam coder because there is no generic way on instanciating a beam coder
---
.../translation/helpers/EncoderHelpers.java | 68 +++++++---------------
.../structuredstreaming/utils/EncodersTest.java | 2 +-
2 files changed, 21 insertions(+), 49 deletions(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index 0751c4c..3f7c102 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -18,7 +18,6 @@
package org.apache.beam.runners.spark.structuredstreaming.translation.helpers;
import static org.apache.spark.sql.types.DataTypes.BinaryType;
-import static scala.compat.java8.JFunction.func;
import java.io.ByteArrayInputStream;
import java.util.ArrayList;
@@ -26,7 +25,6 @@ import java.util.List;
import java.util.Objects;
import org.apache.beam.runners.spark.structuredstreaming.translation.SchemaHelpers;
import org.apache.beam.sdk.coders.Coder;
-import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.spark.sql.Encoder;
@@ -92,17 +90,17 @@ public class EncoderHelpers {
*/
/** A way to construct encoders using generic serializers. */
- public static <T> Encoder<T> fromBeamCoder(Class<? extends Coder<T>> coderClass/*, Class<T> claz*/){
+ public static <T> Encoder<T> fromBeamCoder(Coder<T> beamCoder/*, Class<T> claz*/){
List<Expression> serialiserList = new ArrayList<>();
Class<T> claz = (Class<T>) Object.class;
- serialiserList.add(new EncodeUsingBeamCoder<>(new BoundReference(0, new ObjectType(claz), true), (Class<Coder<T>>)coderClass));
+ serialiserList.add(new EncodeUsingBeamCoder<>(new BoundReference(0, new ObjectType(claz), true), beamCoder));
ClassTag<T> classTag = ClassTag$.MODULE$.apply(claz);
return new ExpressionEncoder<>(
SchemaHelpers.binarySchema(),
false,
JavaConversions.collectionAsScalaIterable(serialiserList).toSeq(),
- new DecodeUsingBeamCoder<>(new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, (Class<Coder<T>>)coderClass),
+ new DecodeUsingBeamCoder<>(new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType), classTag, beamCoder),
classTag);
/*
@@ -125,11 +123,11 @@ public class EncoderHelpers {
public static class EncodeUsingBeamCoder<T> extends UnaryExpression implements NonSQLExpression {
private Expression child;
- private Class<Coder<T>> coderClass;
+ private Coder<T> beamCoder;
- public EncodeUsingBeamCoder(Expression child, Class<Coder<T>> coderClass) {
+ public EncodeUsingBeamCoder(Expression child, Coder<T> beamCoder) {
this.child = child;
- this.coderClass = coderClass;
+ this.beamCoder = beamCoder;
}
@Override public Expression child() {
@@ -138,7 +136,7 @@ public class EncoderHelpers {
@Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
// Code to serialize.
- String beamCoder = lazyInitBeamCoder(ctx, coderClass);
+ String accessCode = ctx.addReferenceObj("beamCoder", beamCoder, beamCoder.getClass().getName());
ExprCode input = child.genCode(ctx);
/*
@@ -172,7 +170,7 @@ public class EncoderHelpers {
args.add(ev.value());
args.add(input.isNull());
args.add(ev.value());
- args.add(beamCoder);
+ args.add(accessCode);
args.add(input.value());
args.add(ev.value());
Block code = (new Block.BlockHelper(sc))
@@ -191,7 +189,7 @@ public class EncoderHelpers {
case 0:
return child;
case 1:
- return coderClass;
+ return beamCoder;
default:
throw new ArrayIndexOutOfBoundsException("productElement out of bounds");
}
@@ -213,11 +211,11 @@ public class EncoderHelpers {
return false;
}
EncodeUsingBeamCoder<?> that = (EncodeUsingBeamCoder<?>) o;
- return coderClass.equals(that.coderClass);
+ return beamCoder.equals(that.beamCoder);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), coderClass);
+ return Objects.hash(super.hashCode(), beamCoder);
}
}
@@ -249,12 +247,12 @@ public class EncoderHelpers {
private Expression child;
private ClassTag<T> classTag;
- private Class<Coder<T>> coderClass;
+ private Coder<T> beamCoder;
- public DecodeUsingBeamCoder(Expression child, ClassTag<T> classTag, Class<Coder<T>> coderClass) {
+ public DecodeUsingBeamCoder(Expression child, ClassTag<T> classTag, Coder<T> beamCoder) {
this.child = child;
this.classTag = classTag;
- this.coderClass = coderClass;
+ this.beamCoder = beamCoder;
}
@Override public Expression child() {
@@ -263,7 +261,7 @@ public class EncoderHelpers {
@Override public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
// Code to deserialize.
- String beamCoder = lazyInitBeamCoder(ctx, coderClass);
+ String accessCode = ctx.addReferenceObj("beamCoder", beamCoder, beamCoder.getClass().getName());
ExprCode input = child.genCode(ctx);
String javaType = CodeGenerator.javaType(dataType());
@@ -298,7 +296,7 @@ public class EncoderHelpers {
args.add(input.isNull());
args.add(CodeGenerator.defaultValue(dataType(), false));
args.add(javaType);
- args.add(beamCoder);
+ args.add(accessCode);
args.add(input.value());
Block code = (new Block.BlockHelper(sc))
.code(JavaConversions.collectionAsScalaIterable(args).toSeq());
@@ -309,10 +307,9 @@ public class EncoderHelpers {
@Override public Object nullSafeEval(Object input) {
try {
- Coder<T> beamCoder = coderClass.getDeclaredConstructor().newInstance();
return beamCoder.decode(new ByteArrayInputStream((byte[]) input));
} catch (Exception e) {
- throw new IllegalStateException("Error decoding bytes for coder: " + coderClass, e);
+ throw new IllegalStateException("Error decoding bytes for coder: " + beamCoder, e);
}
}
@@ -327,7 +324,7 @@ public class EncoderHelpers {
case 1:
return classTag;
case 2:
- return coderClass;
+ return beamCoder;
default:
throw new ArrayIndexOutOfBoundsException("productElement out of bounds");
}
@@ -349,11 +346,11 @@ public class EncoderHelpers {
return false;
}
DecodeUsingBeamCoder<?> that = (DecodeUsingBeamCoder<?>) o;
- return classTag.equals(that.classTag) && coderClass.equals(that.coderClass);
+ return classTag.equals(that.classTag) && beamCoder.equals(that.beamCoder);
}
@Override public int hashCode() {
- return Objects.hash(super.hashCode(), classTag, coderClass);
+ return Objects.hash(super.hashCode(), classTag, beamCoder);
}
}
/*
@@ -384,30 +381,5 @@ case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T], kryo: B
}
*/
- private static <T> String lazyInitBeamCoder(CodegenContext ctx, Class<Coder<T>> coderClass) {
- String beamCoderInstance = "beamCoder";
- ctx.addImmutableStateIfNotExists(coderClass.getName(), beamCoderInstance, func(v1 -> {
- /*
- CODE GENERATED
- try {
- v1 = coderClass.class.getDeclaredConstructor().newInstance();
- } catch (Exception e) {
- throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
- }
- */
- List<String> parts = new ArrayList<>();
- parts.add("try {");
- parts.add(" = (");
- parts.add(") ");
- parts.add(".class.getDeclaredConstructor().newInstance();} catch (Exception e) {throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
- StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
- List<Object> args = new ArrayList<>();
- args.add(v1);
- args.add(coderClass.getName());
- args.add(coderClass.getName());
- return sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
- }));
- return beamCoderInstance;
- }
}
diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
index 0e38fe1..b3a6273 100644
--- a/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
+++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/structuredstreaming/utils/EncodersTest.java
@@ -24,7 +24,7 @@ public class EncodersTest {
data.add(1);
data.add(2);
data.add(3);
- sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.class));
+ sparkSession.createDataset(data, EncoderHelpers.fromBeamCoder(VarIntCoder.of()));
// sparkSession.createDataset(data, EncoderHelpers.genericEncoder());
}
}