Use LinkedIn spark to build spark runner

PawasChhokra · Apr 9, 2020 · ff0c00d · ff0c00d
1 parent 95f0bd9
commit ff0c00d
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 15 deletions.
diff --git a/runners/spark/build.gradle b/runners/spark/build.gradle
@@ -32,6 +32,18 @@ evaluationDependsOn(":sdks:java:core")
 evaluationDependsOn(":sdks:java:io:hadoop-format")
 evaluationDependsOn(":runners:core-java")
 
+repositories {
+  maven {
+    url "https://artifactory.corp.linkedin.com:8083/artifactory/DDS/"
+  }
+  maven {
+    url "https://artifactory.corp.linkedin.com:8083/artifactory/mintdev-publish-repo/"
+  }
+  maven {
+    url "file://" + System.getProperty("user.home") + "/local-repo"
+  }
+}
+
 configurations {
   validatesRunner
 }
@@ -55,6 +67,9 @@ test {
   }
 }
 
+// Use LinkedIn spark
+def spark_version = "2.3.0.263"
+
 dependencies {
   compile project(path: ":model:pipeline", configuration: "shadow")
   compile project(path: ":sdks:java:core", configuration: "shadow")
@@ -65,10 +80,12 @@ dependencies {
   compile library.java.slf4j_api
   compile library.java.joda_time
   compile library.java.args4j
-  provided library.java.spark_core
-  provided library.java.spark_sql
-  provided library.java.spark_streaming
-  provided library.java.spark_network_common
+
+  compile "com.linkedin.spark:spark-core_2.11:$spark_version"
+  compile "com.linkedin.spark:spark-sql_2.11:$spark_version"
+  compile "com.linkedin.spark:spark-streaming_2.11:$spark_version"
+  compile "com.linkedin.spark:spark-network-common_2.11:$spark_version"
+
   provided library.java.hadoop_common
   provided library.java.commons_io
   provided library.java.hamcrest_core
@@ -235,7 +252,7 @@ task validatesRunner {
   description "Validates Spark runner"
   dependsOn validatesRunnerBatch
   dependsOn validatesRunnerStreaming
-  // It should be uncommented once all "validatesStructuredStreamingRunnerBatch" tests will pass. 
+  // It should be uncommented once all "validatesStructuredStreamingRunnerBatch" tests will pass.
   // Otherwise, it breaks Spark runner ValidatesRunner tests.
   //dependsOn validatesStructuredStreamingRunnerBatch
 }

diff --git a/...org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java b/...org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -36,8 +36,6 @@
 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.types.DataType;
@@ -94,7 +92,7 @@ public Expression child() {
     public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
       String accessCode = ctx.addReferenceObj("coder", coder, coder.getClass().getName());
       ExprCode input = child.genCode(ctx);
-      String javaType = CodeGenerator.javaType(dataType());
+      String javaType = ctx.javaType(dataType());
 
       List<String> parts = new ArrayList<>();
       List<Object> args = new ArrayList<>();
@@ -117,10 +115,12 @@ public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
 
       StringContext sc =
           new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
-      Block code =
-          (new Block.BlockHelper(sc)).code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+      // Block code =
+      //    (new
+      // Block.BlockHelper(sc)).code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+      String code = sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
 
-      return ev.copy(input.code().$plus(code), input.isNull(), ev.value());
+      return ev.copy(input.code() + "\n" + code, input.isNull(), ev.value());
     }
 
     @Override
@@ -203,7 +203,7 @@ public Expression child() {
     public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
       String accessCode = ctx.addReferenceObj("coder", coder, coder.getClass().getName());
       ExprCode input = child.genCode(ctx);
-      String javaType = CodeGenerator.javaType(dataType());
+      String javaType = ctx.javaType(dataType());
 
       List<String> parts = new ArrayList<>();
       List<Object> args = new ArrayList<>();
@@ -228,9 +228,11 @@ public ExprCode doGenCode(CodegenContext ctx, ExprCode ev) {
 
       StringContext sc =
           new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
-      Block code =
-          (new Block.BlockHelper(sc)).code(JavaConversions.collectionAsScalaIterable(args).toSeq());
-      return ev.copy(input.code().$plus(code), input.isNull(), ev.value());
+      // Block code =
+      //    (new
+      // Block.BlockHelper(sc)).code(JavaConversions.collectionAsScalaIterable(args).toSeq());
+      String code = sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
+      return ev.copy(input.code() + "\n" + code, input.isNull(), ev.value());
     }
 
     @Override