Spark 4: Update GHA & Get Tests Running Again (#140)

* Update GHA

* Add missing uses for setting up the JDK

* Add sbt explicitly now

* Bump spark testing version

* Update more to Spark 4, except PySpark ex which uses Iceberg leave that at 3.5

* remove loadsave for Spark3 compilation with PySpark.

* Install proto if needed.

* Fix rm

* Add distutils fix typo

* Setuptools

Author: holdenk

.github/workflows/ci.yml

@@ -3,6 +3,12 @@
 set -ex
 source install_rust_if_needed.sh
 
+if command -v protoc >/dev/null 2>&1; then
+  echo "protoc already installed"
+else
+  sudo apt-get install -y protobuf-compiler
+fi
+
 if [ -z "${SPARK_MAJOR}" ]; then
   echo "Need a spark major version specified."
   exit 1

accelerators/setup_comet.sh

@@ -78,7 +78,7 @@ lazy val core = (project in file("core")) // regular scala code with @native met
     Test / javaOptions ++= specialOptions,
     // 2.4.5 is the highest version we have with the old spark-testing-base deps
     sparkVersion := System.getProperty("sparkVersion", "4.0.0"),
-    sparkTestingVersion := "2.0.1",
+    sparkTestingVersion := "2.1.2",
     // additional libraries
     libraryDependencies ++= Seq(
       "org.apache.spark" %% "spark-core"                % sparkVersion.value % Provided,

build.sbt

@@ -4,10 +4,9 @@
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.sql.Column;
 import org.apache.spark.sql.*;
-import org.apache.spark.sql.SQLContext;
+import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.expressions.Window;
 import org.apache.spark.sql.expressions.WindowSpec;
-import org.apache.spark.sql.hive.HiveContext;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -16,39 +15,23 @@
 
 public class JavaHappyPandas {
 
-  /**
-   * Creates SQLContext with an existing SparkContext.
-   */
-  public static SQLContext sqlContext(JavaSparkContext jsc) {
-    SQLContext sqlContext = new SQLContext(jsc);
-    return sqlContext;
-  }
-
-  /**
-   * Creates HiveContext with an existing SparkContext.
-   */
-  public static HiveContext hiveContext(JavaSparkContext jsc) {
-    HiveContext hiveContext = new HiveContext(jsc);
-    return hiveContext;
-  }
-
   /**
    * Illustrate loading some JSON data.
    */
-  public static Dataset<Row> loadDataSimple(JavaSparkContext jsc, SQLContext sqlContext, String path) {
-    Dataset<Row> df1 = sqlContext.read().json(path);
+  public static Dataset<Row> loadDataSimple(JavaSparkContext jsc, SparkSession session, String path) {
+    Dataset<Row> df1 = session.read().json(path);
 
-    Dataset<Row> df2 = sqlContext.read().format("json").option("samplingRatio", "1.0").load(path);
+    Dataset<Row> df2 = session.read().format("json").option("samplingRatio", "1.0").load(path);
 
     JavaRDD<String> jsonRDD = jsc.textFile(path);
-    Dataset<Row> df3 = sqlContext.read().json(jsonRDD);
+    Dataset<Row> df3 = session.read().json(jsonRDD);
 
     return df1;
   }
 
-  public static Dataset<Row> jsonLoadFromRDD(SQLContext sqlContext, JavaRDD<String> input) {
+  public static Dataset<Row> jsonLoadFromRDD(SparkSession session, JavaRDD<String> input) {
     JavaRDD<String> rdd = input.filter(e -> e.contains("panda"));
-    Dataset<Row> df = sqlContext.read().json(rdd);
+    Dataset<Row> df = session.read().json(rdd);
     return df;
   }
 
@@ -147,10 +130,10 @@ public static Dataset<Row> minMeanSizePerZip(Dataset<Row> pandas) {
   }
 
   public static Dataset<Row> simpleSqlExample(Dataset<Row> pandas) {
-    SQLContext sqlContext = pandas.sqlContext();
+    SparkSession session = SparkSession.builder().getOrCreate();
     pandas.registerTempTable("pandas");
 
-    Dataset<Row> miniPandas = sqlContext.sql("SELECT * FROM pandas WHERE pandaSize < 12");
+    Dataset<Row> miniPandas = session.sql("SELECT * FROM pandas WHERE pandaSize < 12");
     return miniPandas;
   }
 

core/src/main/java/com/highperformancespark/examples/dataframe/JavaHappyPandas.java

@@ -348,18 +348,19 @@ object HappyPandas {
    * Cut the lineage of a DataFrame which has too long a query plan.
    */
   def cutLineage(df: DataFrame): DataFrame = {
-    val sqlCtx = df.sqlContext
+    val session = SparkSession.builder.getOrCreate()
+    import session.implicits._
     //tag::cutLineage[]
     val rdd = df.rdd
     rdd.cache()
-    sqlCtx.createDataFrame(rdd, df.schema)
+    session.createDataFrame(rdd, df.schema)
     //end::cutLineage[]
   }
 
   // Self join
   def selfJoin(df: DataFrame): DataFrame = {
-    val sqlCtx = df.sqlContext
-    import sqlCtx.implicits._
+    val session = SparkSession.builder.getOrCreate()
+    import session.implicits._
     //tag::selfJoin[]
     val joined = df.as("a").join(df.as("b")).where($"a.name" === $"b.name")
     //end::selfJoin[]

core/src/main/scala/com/high-performance-spark-examples/dataframe/HappyPandas.scala

@@ -8,7 +8,7 @@ import org.apache.spark.sql.catalyst.optimizer._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern._
-import org.apache.spark.sql.catalyst.expressions.{And, IsNotNull, NullIntolerant}
+import org.apache.spark.sql.catalyst.expressions.{And, IsNotNull}
 
 object NullabilityFilterOptimizer extends Rule[LogicalPlan] {
 

core/src/main/scala/com/high-performance-spark-examples/dataframe/NullabilityFilterOptimizer.scala

@@ -17,7 +17,6 @@ import com.highperformancespark.examples.dataframe._
 
 object SimplePipeline {
   def constructAndSetParams(df: DataFrame) = {
-    val sqlCtx = df.sqlContext
     //tag::constructSetParams[]
     val hashingTF = new HashingTF()
     hashingTF.setInputCol("input")
@@ -26,7 +25,6 @@ object SimplePipeline {
   }
 
   def constructSimpleTransformer(df: DataFrame) = {
-    val sqlCtx = df.sqlContext
     //tag::simpleTransformer[]
     val hashingTF = new HashingTF()
     // We don't set the output column here so the default output column of
@@ -62,7 +60,6 @@ object SimplePipeline {
   }
 
   def constructSimpleEstimator(df: DataFrame) = {
-    val sqlCtx = df.sqlContext
     //tag::simpleNaiveBayes[]
     val nb = new NaiveBayes()
     nb.setLabelCol("happy")

core/src/main/scala/com/high-performance-spark-examples/ml/SimplePipeline.scala

@@ -4,7 +4,7 @@ set -ex
 
 # Download Spark and iceberg if not present
 SPARK_MAJOR=${SPARK_MAJOR:-"3.5"}
-SPARK_VERSION=${SPARK_VERSION:-"${SPARK_MAJOR}.2"}
+SPARK_VERSION=${SPARK_VERSION:-"${SPARK_MAJOR}.3"}
 SCALA_VERSION=${SCALA_VERSION:-"2.13"}
 HADOOP_VERSION="3"
 SPARK_PATH="$(pwd)/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}"
@@ -13,7 +13,7 @@ if [ "$SCALA_VERSION" = "2.13" ]; then
   SPARK_FILE="spark-${SPARK_VERSION}-bin-hadoop3-scala2.13.tgz"
   SPARK_PATH="$(pwd)/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}-scala2.13"
 fi
-ICEBERG_VERSION=${ICEBERG_VERSION:-"1.6.0"}
+ICEBERG_VERSION=${ICEBERG_VERSION:-"1.9.2"}
 if [ ! -f "${SPARK_FILE}" ]; then
   SPARK_DIST_URL="https://dlcdn.apache.org/spark/spark-${SPARK_VERSION}/${SPARK_FILE}"
   SPARK_ARCHIVE_DIST_URL="https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/${SPARK_FILE}"

env_setup.sh

@@ -42,8 +42,12 @@ function check_fail () {
 
 EXAMPLE_JAR="./core/target/scala-2.13/core-assembly-0.1.0-SNAPSHOT.jar"
 
+pip install setuptools
+
+# Iceberg JAR not yet available for Spark 4.
 if [ ! -f "${EXAMPLE_JAR}" ]; then
-  sbt core/assembly
+  rm ./core/src/main/scala/com/high-performance-spark-examples/dataframe/LoadSave.scala # temp hack no merge in Spark 3.
+  sbt core/assembly -DsparkVersion="${SPARK_VERSION}"
 fi
 
 if [ ! -f "${EXAMPLE_JAR}" ]; then