fixed spark 2.0 incompatibility issue. Dataframe no longer exists in… (…

…sryza#162) * fixed spark 2.0 incompatibility issue. Dataframe no longer exists in Java API, now we use Dataset<Row>. * updating travis ci and maven config for the transition from Spark 1.x to 2.x
askasjoe · Sep 8, 2016 · b75164a · b75164a
1 parent 8cba383
commit b75164a
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 34 deletions.
diff --git a/.travis.yml b/.travis.yml
@@ -6,16 +6,11 @@ script: mvn ${SPARK} ${SCALA} -Dmaven.javadoc.skip=true -q -B -Dgpg.skip=true ve
 env:
   global:
     - MAVEN_OPTS=-Xmx2g
-matrix:
-  include:
-    # Covers Spark 1.3.x
-    - env: SPARK=-Dspark.version=1.3.1
-    # Covers Spark 1.4.x
-    - env: SPARK=-Dspark.version=1.4.1
-    # Covers Spark 1.5.x
-    - env: SPARK=-Dspark.version=1.5.2
-    # Covers Scala 2.11, Spark 1.6.x
-    - env: SCALA=-Pscala-2.11 SPARK=-Dspark.version=1.6.0
+# Use this to test future version of the 2.x Spark environment
+# matrix:
+#  include:
+#    # Covers Spark 2.0.x 
+#    - env: SPARK=-Dspark.version=2.0.0
 cache:
   directories:
     - $HOME/.m2
diff --git a/pom.xml b/pom.xml
@@ -36,10 +36,10 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <java.version>1.8</java.version>
-    <scala.minor.version>2.10</scala.minor.version>
-    <scala.complete.version>${scala.minor.version}.4</scala.complete.version>
-    <spark.version>1.3.1</spark.version>
-    <breeze.version>0.10</breeze.version>
+    <scala.minor.version>2.11</scala.minor.version>
+    <scala.complete.version>${scala.minor.version}.8</scala.complete.version>
+    <spark.version>2.0.0</spark.version>
+    <breeze.version>0.12</breeze.version>
   </properties>
 
   <distributionManagement>
@@ -451,19 +451,12 @@
 
   <profiles>
     <profile>
-      <id>scala-2.11</id>
+      <id>scala-2.10</id>
       <properties>
-        <scala.minor.version>2.11</scala.minor.version>
+        <scala.minor.version>2.10</scala.minor.version>
         <scala.complete.version>${scala.minor.version}.6</scala.complete.version>
       </properties>
     </profile>
-    <profile>
-      <id>spark-1.6</id>
-      <properties>
-        <spark.version>1.6.1</spark.version>
-        <breeze.version>0.12</breeze.version>
-      </properties>
-    </profile>
   </profiles>
 
 </project>
diff --git a/src/main/java/com/cloudera/sparkts/api/java/JavaTimeSeriesRDDFactory.java b/src/main/java/com/cloudera/sparkts/api/java/JavaTimeSeriesRDDFactory.java
@@ -19,8 +19,9 @@
 import org.apache.spark.api.java.JavaPairRDD;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
-import org.apache.spark.sql.DataFrame;
 import org.apache.spark.mllib.linalg.Vector;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
 import scala.Tuple3;
 import scala.reflect.ClassTag$;
 
@@ -88,10 +89,11 @@ public static <K> JavaTimeSeriesRDD<K> timeSeriesRDD(
      */
     public static JavaTimeSeriesRDD<String> timeSeriesRDDFromObservations(
             DateTimeIndex targetIndex,
-            DataFrame df,
+            Dataset<Row> df,
             String tsCol,
             String keyCol,
             String valueCol) {
+
         return JAVA_TIME_SERIES_RDD.javaTimeSeriesRDDFromObservations(
                 targetIndex,
                 df,

diff --git a/src/main/scala/com/cloudera/sparkts/api/java/JavaTimeSeriesRDD.scala b/src/main/scala/com/cloudera/sparkts/api/java/JavaTimeSeriesRDD.scala
@@ -23,7 +23,7 @@ import org.apache.spark.{TaskContext, Partition}
 import org.apache.spark.api.java.{JavaSparkContext, JavaRDD, JavaPairRDD}
 import org.apache.spark.mllib.linalg.Vector
 import org.apache.spark.mllib.linalg.distributed.{RowMatrix, IndexedRowMatrix}
-import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spark.sql.{DataFrame, Dataset, Row, SQLContext}
 import org.apache.spark.util.StatCounter
 
 import scala.reflect.ClassTag
@@ -280,7 +280,7 @@ object JavaTimeSeriesRDD {
    */
   def javaTimeSeriesRDDFromObservations(
      targetIndex: DateTimeIndex,
-     df: DataFrame,
+     df: Dataset[Row],
      tsCol: String,
      keyCol: String,
      valueCol: String): JavaTimeSeriesRDD[String] =

diff --git a/src/test/java/com/cloudera/sparkts/api/java/JavaTimeSeriesRDDFactoryTest.java b/src/test/java/com/cloudera/sparkts/api/java/JavaTimeSeriesRDDFactoryTest.java
@@ -27,7 +27,7 @@
 import org.apache.spark.mllib.linalg.distributed.IndexedRow;
 import org.apache.spark.mllib.linalg.distributed.IndexedRowMatrix;
 import org.apache.spark.mllib.linalg.distributed.RowMatrix;
-import org.apache.spark.sql.DataFrame;
+import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.Row$;
 import org.apache.spark.sql.SQLContext;
@@ -190,8 +190,8 @@ public void testToInstantsDataFrame() {
         JavaTimeSeriesRDD<String> tsRdd = JavaTimeSeriesRDDFactory.timeSeriesRDD(
                 index, rdd);
 
-        DataFrame samplesDF = tsRdd.toInstantsDataFrame(sqlContext);
-        Row[] sampleRows = samplesDF.collect();
+        Dataset<Row> samplesDF = tsRdd.toInstantsDataFrame(sqlContext);
+        Row[] sampleRows = (Row[])samplesDF.collect();
         String[] columnNames = samplesDF.columns();
         String[] columnNamesTail = new String[columnNames.length - 1];
         System.arraycopy(columnNames, 1, columnNamesTail, 0, columnNamesTail.length);
@@ -205,7 +205,7 @@ public void testToInstantsDataFrame() {
           rowFrom(Timestamp.from(start.plusDays(1).toInstant()), untilBy(1, 20, 4)),
           rowFrom(Timestamp.from(start.plusDays(2).toInstant()), untilBy(2, 20, 4)),
           rowFrom(Timestamp.from(start.plusDays(3).toInstant()), untilBy(3, 20, 4))
-        }, sampleRows);
+        }, (Row[])sampleRows);
 
         sc.close();
     }
@@ -337,7 +337,7 @@ public void testTimeSeriesRDDFromObservationsDataFrame() {
         JavaTimeSeriesRDD<String> tsRdd = JavaTimeSeriesRDDFactory.timeSeriesRDD(
                 index, rdd);
 
-        DataFrame obsDF = tsRdd.toObservationsDataFrame(sqlContext, "timestamp", "key", "value");
+        Dataset<Row> obsDF = tsRdd.toObservationsDataFrame(sqlContext, "timestamp", "key", "value");
         JavaTimeSeriesRDD<String> tsRddFromDF = JavaTimeSeriesRDDFactory
                 .timeSeriesRDDFromObservations(
                     index, obsDF, "timestamp", "key", "value");
@@ -348,8 +348,8 @@ public void testTimeSeriesRDDFromObservationsDataFrame() {
                 tsRddFromDF.sortByKey().collect().toArray()
         );
 
-        Row[] df1 = obsDF.collect();
-        Row[] df2 = tsRddFromDF.toObservationsDataFrame(
+        Row[] df1 = (Row[])obsDF.collect();
+        Row[] df2 = (Row[])tsRddFromDF.toObservationsDataFrame(
                 sqlContext, "timestamp", "key", "value").collect();
 
         Comparator<Row> comparator = (r1, r2) -> {