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 <echauc...@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());
}
}
