Use reflection to call SQLContext.createDataFrame

- Spark 1 and Spark 2 are not binary compatible for the method

Author: chtyim

pom.xml

@@ -22,7 +22,7 @@
 
   <groupId>co.cask.hydrator</groupId>
   <artifactId>dynamic-spark</artifactId>
-  <version>2.0.5</version>
+  <version>2.0.6</version>
 
   <properties>
     <!-- properties for script build step that creates the config files for the artifacts -->

src/main/java/co/cask/hydrator/plugin/spark/dynamic/ScalaSparkCompute.java

@@ -48,6 +48,7 @@
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
@@ -207,7 +208,7 @@ public JavaRDD<StructuredRecord> transform(SparkExecutionPluginContext context,
     StructType rowType = DataFrames.toDataType(inputSchema);
     JavaRDD<Row> rowRDD = javaRDD.map(new RecordToRow(rowType));
 
-    Object dataFrame = sqlContext.createDataFrame(rowRDD, rowType);
+    Object dataFrame = createDataFrame(sqlContext, rowRDD, rowType);
     Object result = takeContext ? method.invoke(null, dataFrame, context) : method.invoke(null, dataFrame);
 
     // Convert the DataFrame back to RDD<StructureRecord>
@@ -373,6 +374,20 @@ private boolean isDataFrame(Type type) {
     return false;
   }
 
+  private Object createDataFrame(SQLContext sqlContext, JavaRDD<Row> rdd, StructType type) {
+    try {
+      return sqlContext.getClass()
+        .getMethod("createDataFrame", JavaRDD.class, StructType.class)
+        .invoke(sqlContext, rdd, type);
+    } catch (NoSuchMethodException e) {
+      throw new RuntimeException("Unable to find SQLContext.createDataFrame(JavaRDD, StructType) method", e);
+    } catch (IllegalAccessException e) {
+      throw new RuntimeException("No permission to invoke SQLContext.createDataFrame(JavaRDD, StructType) method", e);
+    } catch (InvocationTargetException e) {
+      throw new RuntimeException("Failed to invoke SQLContext.createDataFrame(JavaRDD, StructType) method", e);
+    }
+  }
+
   /**
    * Configuration object for the plugin
    */