diff --git a/spark-plugin/build.sbt b/spark-plugin/build.sbt index 5715a85a..928e543c 100644 --- a/spark-plugin/build.sbt +++ b/spark-plugin/build.sbt @@ -3,7 +3,7 @@ import sbtassembly.AssemblyPlugin.autoImport._ lazy val versionNum: String = "0.9.10" lazy val scala212 = "2.12.20" -lazy val scala213 = "2.13.16" +lazy val scala213 = "2.13.18" lazy val supportedScalaVersions = List(scala212, scala213) lazy val dataflint = project @@ -20,7 +20,8 @@ lazy val dataflint = project example_3_5_1, example_3_4_1_remote, example_4_0_1, - example_4_1_0 + example_4_1_0, + example_4_2_0 ).settings( crossScalaVersions := Nil, // Aggregate project version must be Nil, see docs: https://www.scala-sbt.org/1.x/docs/Cross-Build.html publish / skip := true @@ -167,13 +168,14 @@ lazy val pluginspark4 = (project in file("pluginspark4")) // Include resources from plugin directory for static UI files Compile / unmanagedResourceDirectories += (plugin / Compile / resourceDirectory).value, - // Test dependencies — Spark 4.0.1 + scalatest. Mirrors pluginspark3 so we can run the + // Test dependencies — Spark 4.2.0 + scalatest. Mirrors pluginspark3 so we can run the // same regression suites against the Spark 4 surface (cross-version validation). + // 4.2.0 ships Jetty 12 (ee10); DataflintJettyUtilsSpec covers that path. // Requires the launching JVM to be Java 17+ since Spark 4 won't run on Java 8/11. libraryDependencies += "org.scalatest" %% "scalatest-funsuite" % "3.2.17" % Test, libraryDependencies += "org.scalatest" %% "scalatest-shouldmatchers" % "3.2.17" % Test, - libraryDependencies += "org.apache.spark" %% "spark-core" % "4.0.1" % Test, - libraryDependencies += "org.apache.spark" %% "spark-sql" % "4.0.1" % Test, + libraryDependencies += "org.apache.spark" %% "spark-core" % "4.2.0" % Test, + libraryDependencies += "org.apache.spark" %% "spark-sql" % "4.2.0" % Test, // Share version-portable test sources with pluginspark3. Most pluginspark3 specs // depend on Spark-3-only internals (Dataset constructor, PythonMapInArrowExec, etc.) @@ -355,4 +357,20 @@ lazy val example_4_1_0 = (project in file("example_4_1_0")) libraryDependencies += "org.apache.spark" % "spark-core_2.13" % "4.1.0", libraryDependencies += "org.apache.spark" % "spark-sql_2.13" % "4.1.0", publish / skip := true - ).dependsOn(pluginspark4) \ No newline at end of file + ).dependsOn(pluginspark4) + +lazy val example_4_2_0 = (project in file("example_4_2_0")) + .settings( + name := "DataflintSparkExample420", + organization := "io.dataflint", + scalaVersion := scala213, + crossScalaVersions := List(scala213), // Only Scala 2.13 for Spark 4.x + // Spark 4.2.0 requires Scala 2.13.18+ (MurmurHash3.caseClassHash) + libraryDependencies += "org.apache.spark" % "spark-core_2.13" % "4.2.0", + libraryDependencies += "org.apache.spark" % "spark-sql_2.13" % "4.2.0", + // Fork so the run classpath uses this project's Scala, not sbt's. + // Keep CWD at the build root so ./test_data/... resolves like the other examples. + run / fork := true, + run / baseDirectory := (LocalRootProject / baseDirectory).value, + publish / skip := true + ).dependsOn(pluginspark4) diff --git a/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala b/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala new file mode 100644 index 00000000..abba770d --- /dev/null +++ b/spark-plugin/example_4_2_0/src/main/scala/io/dataflint/example/ShakespeareSpark420.scala @@ -0,0 +1,51 @@ +package io.dataflint.example + +import org.apache.spark.sql.{DataFrame, SparkSession} +import org.apache.spark.sql.functions._ + +object ShakespeareSpark420 extends App { + def df(spark: SparkSession): DataFrame = spark.read + .format("csv") + .option("sep", ";") + .option("inferSchema", true) + .load("./test_data/will_play_text.csv") + .toDF("line_id", "play_name", "speech_number", "line_number", "speaker", "text_entry") + .repartition(1000) + + val spark = SparkSession + .builder() + .appName("Shakespeare Statistics") + .config("spark.plugins", "io.dataflint.spark.SparkDataflintPlugin") + .config("spark.dataflint.telemetry.enabled", false) + .config("spark.ui.port", "10000") + .master("local[*]") + .getOrCreate() + + import spark.implicits._ + + val shakespeareText = df(spark) + + shakespeareText.printSchema() + + val count = shakespeareText.count() + println(s"number of records : $count") + + val uniqueSpeakers = shakespeareText.select($"speaker").distinct().count() + println(s"number of unique speakers : $uniqueSpeakers") + + val uniqueWords = shakespeareText.select(explode(split($"text_entry", " "))).distinct().count() + println(s"number of unique words : $uniqueWords") + + println("DataFlint UI ready at: http://localhost:10000/dataflint/") + println("Press Enter to stop...") + + // Interactive: wait for Enter. Non-interactive: optional keep-alive for smoke tests. + if (System.console() != null) { + scala.io.StdIn.readLine() + } else { + sys.env.get("DATAFLINT_KEEP_UI_SECONDS").foreach { secs => + Thread.sleep(secs.toLong * 1000) + } + } + spark.stop() +} diff --git a/spark-plugin/pluginspark4/src/main/scala/org/apache/spark/dataflint/api/DataflintJettyUtils.scala b/spark-plugin/pluginspark4/src/main/scala/org/apache/spark/dataflint/api/DataflintJettyUtils.scala index 7f7994de..b51f3e20 100644 --- a/spark-plugin/pluginspark4/src/main/scala/org/apache/spark/dataflint/api/DataflintJettyUtils.scala +++ b/spark-plugin/pluginspark4/src/main/scala/org/apache/spark/dataflint/api/DataflintJettyUtils.scala @@ -15,25 +15,44 @@ object DataflintJettyUtils { // copy of createStaticHandler in core/src/main/scala/org/apache/spark/ui/JettyUtils.scala // only difference is we are loading the resources from this class loader which might be different from the spark one - // with use reflection to support both org.sparkproject.jetty.servlet and org.eclipse.jetty - // in spark source code + // we use reflection to support the different Jetty package layouts Spark has shipped: + // - org.sparkproject.jetty.ee10.servlet (Spark 4.2+, shaded Jetty 12 / EE10) + // - org.sparkproject.jetty.servlet (Spark 3.x / 4.0 / 4.1, shaded Jetty 9-11) + // - org.eclipse.jetty[.ee10].servlet (unshaded fallbacks) private def createStaticHandler(resourceBase: String, path: String): Any = { - // Try to load classes from both packages + // Try to load the class from every Jetty package layout Spark has used, most-recent first. def getClassForName(className: String): Class[_] = { - try { - Class.forName(s"org.sparkproject.jetty.servlet.$className") - } catch { - case _: ClassNotFoundException => Class.forName(s"org.eclipse.jetty.servlet.$className") - } + val candidates = Seq( + s"org.sparkproject.jetty.ee10.servlet.$className", + s"org.sparkproject.jetty.servlet.$className", + s"org.eclipse.jetty.ee10.servlet.$className", + s"org.eclipse.jetty.servlet.$className" + ) + candidates.iterator + .flatMap { name => + try Some(Class.forName(name)) + catch { case _: ClassNotFoundException => None } + } + .nextOption() + .getOrElse(throw new ClassNotFoundException( + s"Could not load Jetty class '$className' from any known package: ${candidates.mkString(", ")}")) } val servletContextHandlerClass = getClassForName("ServletContextHandler") val defaultServletClass = getClassForName("DefaultServlet") val servletHolderClass = getClassForName("ServletHolder") + // Jetty 12 (EE10) renamed the DefaultServlet resource init-param from "resourceBase" to + // "baseResource" and moved the gzip init-param under the ee10 package namespace. + val isEe10 = servletContextHandlerClass.getName.contains(".ee10.") + val gzipInitParam = + if (isEe10) "org.eclipse.jetty.ee10.servlet.Default.gzip" + else "org.eclipse.jetty.servlet.Default.gzip" + val resourceBaseInitParam = if (isEe10) "baseResource" else "resourceBase" + val contextHandler = servletContextHandlerClass.getDeclaredConstructor().newInstance() val setInitParameterMethod = contextHandler.getClass.getMethod("setInitParameter", classOf[String], classOf[String]) - setInitParameterMethod.invoke(contextHandler, "org.eclipse.jetty.servlet.Default.gzip", "false") + setInitParameterMethod.invoke(contextHandler, gzipInitParam, "false") val staticHandler = defaultServletClass.getDeclaredConstructor().newInstance() val servletHolderConstructor = servletHolderClass.getConstructor(classOf[Servlet]) @@ -42,7 +61,7 @@ object DataflintJettyUtils { Option(this.getClass.getClassLoader.getResource(resourceBase)) match { case Some(res) => val setInitParameterMethodForHolder = holder.getClass.getMethod("setInitParameter", classOf[String], classOf[String]) - setInitParameterMethodForHolder.invoke(holder, "resourceBase", res.toString) + setInitParameterMethodForHolder.invoke(holder, resourceBaseInitParam, res.toString) case None => throw new Exception("Could not find resource path for Web UI: " + resourceBase) } diff --git a/spark-plugin/pluginspark4/src/test/scala/org/apache/spark/dataflint/api/DataflintJettyUtilsSpec.scala b/spark-plugin/pluginspark4/src/test/scala/org/apache/spark/dataflint/api/DataflintJettyUtilsSpec.scala new file mode 100644 index 00000000..b9a545ca --- /dev/null +++ b/spark-plugin/pluginspark4/src/test/scala/org/apache/spark/dataflint/api/DataflintJettyUtilsSpec.scala @@ -0,0 +1,60 @@ +package org.apache.spark.dataflint.api + +import java.net.HttpURLConnection +import java.net.URI + +import org.apache.spark.sql.SparkSession +import org.scalatest.BeforeAndAfterAll +import org.scalatest.funsuite.AnyFunSuite +import org.scalatest.matchers.should.Matchers + +/** + * Regression test for Spark 4.2 Jetty 12 (ee10) static-handler support. + * + * Without the ee10 reflective lookup / baseResource init-param fix, + * DataflintJettyUtils.createStaticHandler throws ClassNotFoundException and the + * DataFlint UI returns HTTP 500. + */ +class DataflintJettyUtilsSpec extends AnyFunSuite with Matchers with BeforeAndAfterAll { + + private var spark: SparkSession = _ + + override def beforeAll(): Unit = { + spark = SparkSession.builder() + .master("local[1]") + .appName("DataflintJettyUtilsSpec") + .config("spark.plugins", "io.dataflint.spark.SparkDataflintPlugin") + .config("spark.dataflint.telemetry.enabled", "false") + .config("spark.ui.enabled", "true") + .config("spark.ui.port", "0") + .getOrCreate() + } + + override def afterAll(): Unit = { + if (spark != null) spark.stop() + } + + test("Spark 4.2 classpath exposes Jetty 12 ee10 servlet classes") { + noException should be thrownBy { + Class.forName("org.sparkproject.jetty.ee10.servlet.ServletContextHandler") + Class.forName("org.sparkproject.jetty.ee10.servlet.DefaultServlet") + Class.forName("org.sparkproject.jetty.ee10.servlet.ServletHolder") + } + } + + test("DataFlint static UI handler serves index.html over Jetty 12 ee10") { + val ui = spark.sparkContext.ui.getOrElse( + fail("Spark UI was not started; cannot verify static handler") + ) + val url = s"${ui.webUrl.stripSuffix("/")}/dataflint/index.html" + val connection = URI.create(url).toURL.openConnection().asInstanceOf[HttpURLConnection] + try { + connection.setRequestMethod("GET") + connection.setConnectTimeout(10000) + connection.setReadTimeout(10000) + connection.getResponseCode shouldBe 200 + } finally { + connection.disconnect() + } + } +} diff --git a/spark-plugin/utils/download-spark-versions.sh b/spark-plugin/utils/download-spark-versions.sh index cf1906ca..097b78ec 100755 --- a/spark-plugin/utils/download-spark-versions.sh +++ b/spark-plugin/utils/download-spark-versions.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash # -# Downloads Apache Spark distributions for all supported versions (3.0.x – 4.1.x). +# Downloads Apache Spark distributions for all supported versions (3.0.x – 4.2.x). # Each version is extracted into .spark-versions// under this directory. # # Usage: @@ -26,6 +26,7 @@ ALL_VERSIONS=" 3.5.8:hadoop3 4.0.2:hadoop3 4.1.0:hadoop3 +4.2.0:hadoop3 " get_hadoop_suffix() {