make metrics smaller when there is a lot of data

Author: menishmueli

spark-plugin/build.sbt

@@ -1,28 +1,32 @@
 import xerial.sbt.Sonatype._
+import sbtassembly.AssemblyPlugin.autoImport._
 
 lazy val versionNum: String = "0.5.1"
-lazy val scala212 = "2.12.18"
-lazy val scala213 = "2.13.12"
+lazy val scala212 = "2.12.20"
+lazy val scala213 = "2.13.16"
 lazy val supportedScalaVersions = List(scala212, scala213)
 
 lazy val dataflint = project
   .in(file("."))
   .aggregate(
     plugin,
+    pluginspark3,
+    pluginspark4,
     example_3_1_3,
     example_3_2_4,
     example_3_3_3,
     example_3_4_1,
     example_3_5_1,
-    example_3_4_1_remote
+    example_3_4_1_remote,
+    example_4_0_1
   ).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
   )
 
 lazy val plugin = (project in file("plugin"))
   .settings(
-    name := "spark",
+    name := "dataflint-common",
     organization := "io.dataflint",
     crossScalaVersions := supportedScalaVersions,
     version      := (if (git.gitCurrentTags.value.exists(_.startsWith("v"))) {
@@ -34,7 +38,76 @@ lazy val plugin = (project in file("plugin"))
     libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.5.1"  % "provided",
     libraryDependencies +=  "com.amazonaws" % "aws-java-sdk-s3" % "1.12.470" % "provided",
     libraryDependencies += "org.apache.iceberg" %% "iceberg-spark-runtime-3.5" % "1.5.0" % "provided",
+    publishTo := Some(Resolver.file("local-common", file("target/local-repo"))), // Publish locally only, not to remote repositories
+  )
+
+lazy val pluginspark3 = (project in file("pluginspark3"))
+  .enablePlugins(AssemblyPlugin)
+  .settings(
+    name := "spark", // Keep the name as "spark" for compatibility with existing versions
+    organization := "io.dataflint",
+    crossScalaVersions := supportedScalaVersions,
+    version      := (if (git.gitCurrentTags.value.exists(_.startsWith("v"))) {
+      versionNum
+    } else {
+      versionNum + "-SNAPSHOT"
+    }),
+    libraryDependencies += "org.apache.spark" %% "spark-core" % "3.5.1" % "provided",
+    libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.5.1"  % "provided",
+    libraryDependencies +=  "com.amazonaws" % "aws-java-sdk-s3" % "1.12.470" % "provided",
+    libraryDependencies += "org.apache.iceberg" %% "iceberg-spark-runtime-3.5" % "1.5.0" % "provided",
+    
+    // Assembly configuration to create fat JAR with common code
+    assembly / assemblyJarName := s"${name.value}_${scalaBinaryVersion.value}-${version.value}.jar",
+    assembly / assemblyMergeStrategy := {
+      case PathList("META-INF", "services", xs @ _*) => MergeStrategy.concat
+      case PathList("META-INF", xs @ _*) => MergeStrategy.discard
+      case "application.conf" => MergeStrategy.concat
+      case "reference.conf" => MergeStrategy.concat
+      case _ => MergeStrategy.first
+    },
+    
+    // Publish the assembled JAR instead of the regular JAR
+    Compile / packageBin := assembly.value,
     publishTo := sonatypePublishToBundle.value
+  ).dependsOn(plugin % "compile->compile;test->test")
+
+lazy val pluginspark4 = (project in file("pluginspark4"))
+  .enablePlugins(AssemblyPlugin)
+  .settings(
+    name := "dataflint-spark4",
+    organization := "io.dataflint",
+    scalaVersion := scala213,
+    crossScalaVersions := List(scala213), // Only Scala 2.13 for Spark 4.x
+    version      := (if (git.gitCurrentTags.value.exists(_.startsWith("v"))) {
+      versionNum
+    } else {
+      versionNum + "-SNAPSHOT"
+    }),
+    libraryDependencies += "org.apache.spark" %% "spark-core" % "4.0.1" % "provided",
+    libraryDependencies += "org.apache.spark" %% "spark-sql" % "4.0.1"  % "provided",
+    libraryDependencies +=  "com.amazonaws" % "aws-java-sdk-s3" % "1.12.470" % "provided",
+    libraryDependencies += "org.apache.iceberg" %% "iceberg-spark-runtime-3.5" % "1.5.0" % "provided",
+    
+    // Assembly configuration to create fat JAR with common code
+    assembly / assemblyJarName := s"${name.value}_${scalaBinaryVersion.value}-${version.value}.jar",
+    assembly / assemblyMergeStrategy := {
+      case PathList("META-INF", "services", xs @ _*) => MergeStrategy.concat
+      case PathList("META-INF", xs @ _*) => MergeStrategy.discard
+      case "application.conf" => MergeStrategy.concat
+      case "reference.conf" => MergeStrategy.concat
+      case _ => MergeStrategy.first
+    },
+    
+    // Publish the assembled JAR instead of the regular JAR
+    Compile / packageBin := assembly.value,
+    publishTo := sonatypePublishToBundle.value,
+    
+    // Include source from plugin directory for self-contained build
+    Compile / unmanagedSourceDirectories += (plugin / Compile / sourceDirectory).value / "scala",
+    
+    // Include resources from plugin directory for static UI files
+    Compile / unmanagedResourceDirectories += (plugin / Compile / resourceDirectory).value
   )
 
 lazy val example_3_1_3 = (project in file("example_3_1_3"))
@@ -45,7 +118,7 @@ lazy val example_3_1_3 = (project in file("example_3_1_3"))
     libraryDependencies += "org.apache.spark" %% "spark-core" % "3.1.3",
     libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.1.3",
     publish / skip := true
-  ).dependsOn(plugin)
+  ).dependsOn(pluginspark3)
 
 lazy val example_3_2_4 = (project in file("example_3_2_4"))
   .settings(
@@ -56,7 +129,7 @@ lazy val example_3_2_4 = (project in file("example_3_2_4"))
     libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.2.4",
     libraryDependencies +=  "com.amazonaws" % "aws-java-sdk-s3" % "1.12.470",
     publish / skip := true
-  ).dependsOn(plugin)
+  ).dependsOn(pluginspark3)
 
 lazy val example_3_3_3 = (project in file("example_3_3_3"))
   .settings(
@@ -69,7 +142,7 @@ lazy val example_3_3_3 = (project in file("example_3_3_3"))
     libraryDependencies += "org.apache.iceberg" %% "iceberg-spark-runtime-3.3" % "1.5.0",
     libraryDependencies += "org.scala-lang.modules" %% "scala-collection-compat" % "2.11.0",
     publish / skip := true
-  ).dependsOn(plugin)
+  ).dependsOn(pluginspark3)
 
 lazy val example_3_4_1 = (project in file("example_3_4_1"))
   .settings(
@@ -81,7 +154,7 @@ lazy val example_3_4_1 = (project in file("example_3_4_1"))
     libraryDependencies +=  "com.amazonaws" % "aws-java-sdk-s3" % "1.12.470",
     libraryDependencies += "org.apache.iceberg" %% "iceberg-spark-runtime-3.4" % "1.5.0",
     publish / skip := true
-  ).dependsOn(plugin)
+  ).dependsOn(pluginspark3)
 
 lazy val example_3_5_1 = (project in file("example_3_5_1"))
   .settings(
@@ -99,7 +172,7 @@ lazy val example_3_5_1 = (project in file("example_3_5_1"))
     libraryDependencies += "org.scalatest" %% "scalatest" % "3.2.17",
     libraryDependencies += "org.apache.datafusion" % "comet-spark-spark3.5_2.12" % "0.4.0",
     publish / skip := true
-  ).dependsOn(plugin)
+  ).dependsOn(pluginspark3)
 
 lazy val example_3_4_1_remote = (project in file("example_3_4_1_remote"))
   .settings(
@@ -112,3 +185,14 @@ lazy val example_3_4_1_remote = (project in file("example_3_4_1_remote"))
     publish / skip := true
   ).dependsOn()
 
+lazy val example_4_0_1 = (project in file("example_4_0_1"))
+  .settings(
+    name := "DataflintSparkExample401",
+    organization := "io.dataflint",
+    scalaVersion := scala213,
+    crossScalaVersions := List(scala213), // Only Scala 2.13 for Spark 4.x
+    // there is no scala 2.12 version so we need to force 2.13 to make it compile
+    libraryDependencies += "org.apache.spark" % "spark-core_2.13" % "4.0.1",
+    libraryDependencies += "org.apache.spark" % "spark-sql_2.13" % "4.0.1",
+    publish / skip := true
+  ).dependsOn(pluginspark4)

spark-plugin/clean-and-setup.sh

@@ -0,0 +1,43 @@
+#!/bin/bash
+
+# Clean and setup script for DataFlint Spark plugin development
+# This resolves cross-version dependency conflicts in IntelliJ IDEA
+
+echo "🧹 Cleaning local ivy cache and build artifacts..."
+
+# Remove all local dataflint artifacts
+rm -rf ~/.ivy2/local/io.dataflint/
+
+# Clean all target directories
+find . -name "target" -type d -exec rm -rf {} + 2>/dev/null || true
+
+echo "📦 Publishing required versions..."
+
+# Publish Scala 2.12 version for Spark 3.x projects
+echo "Publishing Scala 2.12 version..."
+sbt plugin/publishLocal
+
+# Publish Scala 2.13 version for Spark 4.x projects  
+echo "Publishing Scala 2.13 version..."
+sbt ++2.13.16 plugin/publishLocal
+
+echo "🔨 Building fat JARs..."
+
+# Build Spark 3 fat JAR
+echo "Building Spark 3 fat JAR..."
+sbt pluginspark3/assembly
+
+# Build Spark 4 fat JAR
+echo "Building Spark 4 fat JAR..."
+sbt ++2.13.16 pluginspark4/assembly
+
+echo "✅ Setup complete!"
+echo ""
+echo "📋 Next steps:"
+echo "1. Refresh your IntelliJ IDEA project (File -> Reload Gradle Project or similar)"
+echo "2. If you still get conflicts, try: File -> Invalidate Caches and Restart"
+echo ""
+echo "📦 Fat JARs created:"
+echo "- Spark 3.x: pluginspark3/target/scala-2.12/dataflint-spark3_2.12-0.5.1-SNAPSHOT.jar"
+echo "- Spark 4.x: pluginspark4/target/scala-2.13/dataflint-spark4_2.13-0.5.1-SNAPSHOT.jar"
+

spark-plugin/example_4_0_1/src/main/scala/io/dataflint/example/ShakespeareSpark401.scala

@@ -0,0 +1,42 @@
+package io.dataflint.example
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.functions._
+
+object ShakespeareSpark401 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")
+
+  scala.io.StdIn.readLine()
+  spark.stop()
+}

spark-plugin/plugin/src/main/resources/META-INF/services/org.apache.spark.status.AppHistoryServerPlugin

@@ -1,7 +1,7 @@
 package org.apache.spark.dataflint
 
 import org.apache.spark.SparkContext
-import org.apache.spark.dataflint.api.{DataFlintTab, DataflintApplicationInfoPage, DataflintCachedStoragePage, DataflintIcebergPage, DataflintJettyUtils, DataflintSQLMetricsPage, DataflintSQLPlanPage, DataflintSQLStagesRddPage}
+import org.apache.spark.dataflint.api.DataflintPageFactory
 import org.apache.spark.dataflint.listener.{DataflintEnvironmentInfo, DataflintEnvironmentInfoEvent}
 import org.apache.spark.dataflint.iceberg.ClassLoaderChecker
 import org.apache.spark.dataflint.iceberg.ClassLoaderChecker.isMetricLoaderInRightClassLoader
@@ -14,13 +14,13 @@ import org.apache.spark.sql.execution.ui.SQLAppStatusListener
 import org.apache.spark.status.{ElementTrackingStore, LiveRDDsListener}
 import org.apache.spark.ui.SparkUI
 
-class DataflintSparkUIInstaller extends Logging {
-  def install(context: SparkContext): String = {
+class DataflintSparkUICommonInstaller extends Logging {
+  def install(context: SparkContext, pageFactory: DataflintPageFactory): String = {
     if(context.ui.isEmpty) {
       logWarning("No UI detected, skipping installation...")
       return ""
     }
-    val isDataFlintAlreadyInstalled = context.ui.get.getTabs.exists(_.name == "DataFlint")
+    val isDataFlintAlreadyInstalled = pageFactory.getTabs(context.ui.get).exists(_.name == "DataFlint")
     if(isDataFlintAlreadyInstalled){
       logInfo("DataFlint UI is already installed, skipping installation...")
       return context.ui.get.webUrl
@@ -95,35 +95,46 @@ class DataflintSparkUIInstaller extends Logging {
         logWarning("Could not add DataFlint Listeners to listener bus", e)
     }
 
-    loadUI(context.ui.get, sqlListener)
+    loadUI(context.ui.get, pageFactory, sqlListener)
   }
 
-  def loadUI(ui: SparkUI, sqlListener: () => Option[SQLAppStatusListener] = () => None): String = {
-    val isDataFlintAlreadyInstalled = ui.getTabs.exists(_.name == "DataFlint")
+  def loadUI(ui: SparkUI, pageFactory: DataflintPageFactory, sqlListener: () => Option[SQLAppStatusListener] = () => None): String = {
+    val isDataFlintAlreadyInstalled = pageFactory.getTabs(ui).exists(_.name == "DataFlint")
     if (isDataFlintAlreadyInstalled) {
       logInfo("DataFlint UI is already installed, skipping installation...")
       return ui.webUrl
     }
-    DataflintJettyUtils.addStaticHandler(ui, "io/dataflint/spark/static/ui", ui.basePath + "/dataflint")
+    pageFactory.addStaticHandler(ui, "io/dataflint/spark/static/ui", ui.basePath + "/dataflint")
     val dataflintStore = new DataflintStore(store = ui.store.store)
-    val tab = new DataFlintTab(ui)
-    tab.attachPage(new DataflintSQLPlanPage(ui, dataflintStore, sqlListener))
-    tab.attachPage(new DataflintSQLMetricsPage(ui, sqlListener))
-    tab.attachPage(new DataflintSQLStagesRddPage(ui))
-    tab.attachPage(new DataflintApplicationInfoPage(ui, dataflintStore))
-    tab.attachPage(new DataflintIcebergPage(ui, dataflintStore))
-    tab.attachPage(new DataflintCachedStoragePage(ui,  dataflintStore))
+    val tab = pageFactory.createDataFlintTab(ui)
+    tab.attachPage(pageFactory.createSQLPlanPage(ui, dataflintStore, sqlListener))
+    tab.attachPage(pageFactory.createSQLMetricsPage(ui, sqlListener))
+    tab.attachPage(pageFactory.createSQLStagesRddPage(ui))
+    tab.attachPage(pageFactory.createApplicationInfoPage(ui, dataflintStore))
+    tab.attachPage(pageFactory.createIcebergPage(ui, dataflintStore))
+    tab.attachPage(pageFactory.createCachedStoragePage(ui, dataflintStore))
     ui.attachTab(tab)
     ui.webUrl
   }
 
 }
-object DataflintSparkUILoader {
+
+object DataflintSparkUICommonLoader {
+  
+  def install(context: SparkContext, pageFactory: DataflintPageFactory): String = {
+    new DataflintSparkUICommonInstaller().install(context, pageFactory)
+  }
+
+  def loadUI(ui: SparkUI, pageFactory: DataflintPageFactory): String = {
+    new DataflintSparkUICommonInstaller().loadUI(ui, pageFactory)
+  }
+  
+  // Backward compatibility methods - these will be overridden in version-specific implementations
   def install(context: SparkContext): String = {
-    new DataflintSparkUIInstaller().install(context)
+    throw new UnsupportedOperationException("This method requires a version-specific implementation. Use pluginspark3 or pluginspark4.")
   }
 
   def loadUI(ui: SparkUI): String = {
-    new DataflintSparkUIInstaller().loadUI(ui)
+    throw new UnsupportedOperationException("This method requires a version-specific implementation. Use pluginspark3 or pluginspark4.")
   }
-}
+}

spark-plugin/plugin/src/main/scala/org/apache/spark/dataflint/DataflintSparkUILoader.scala renamed to spark-plugin/plugin/src/main/scala/org/apache/spark/dataflint/DataflintSparkUICommonLoader.scala

@@ -0,0 +1,30 @@
+package org.apache.spark.dataflint.api
+
+import org.apache.spark.dataflint.listener.DataflintStore
+import org.apache.spark.sql.execution.ui.SQLAppStatusListener
+import org.apache.spark.ui.{SparkUI, WebUIPage, WebUITab}
+
+/**
+ * Abstract factory for creating Dataflint UI components.
+ * This allows version-specific implementations for different Spark versions.
+ */
+abstract class DataflintPageFactory {
+  
+  def createDataFlintTab(ui: SparkUI): WebUITab
+  
+  def createApplicationInfoPage(ui: SparkUI, dataflintStore: DataflintStore): WebUIPage
+  
+  def createCachedStoragePage(ui: SparkUI, dataflintStore: DataflintStore): WebUIPage
+  
+  def createIcebergPage(ui: SparkUI, dataflintStore: DataflintStore): WebUIPage
+  
+  def createSQLMetricsPage(ui: SparkUI, sqlListener: () => Option[SQLAppStatusListener]): WebUIPage
+  
+  def createSQLPlanPage(ui: SparkUI, dataflintStore: DataflintStore, sqlListener: () => Option[SQLAppStatusListener]): WebUIPage
+  
+  def createSQLStagesRddPage(ui: SparkUI): WebUIPage
+  
+  def addStaticHandler(ui: SparkUI, resourceBase: String, contextPath: String): Unit
+  
+  def getTabs(ui: SparkUI): Seq[WebUITab]
+}

spark-plugin/plugin/src/main/scala/org/apache/spark/dataflint/api/DataflintPageFactory.scala

@@ -0,0 +1 @@
+org.apache.spark.deploy.history.DataFlintHistoryServerPlugin

spark-plugin/pluginspark3/src/main/resources/META-INF/services/org.apache.spark.status.AppHistoryServerPlugin

@@ -1,7 +1,7 @@
 package io.dataflint.spark
 
 import org.apache.spark.SparkContext
-import org.apache.spark.dataflint.DataflintSparkUILoader
+import org.apache.spark.dataflint.{DataflintSparkUICommonLoader, DataflintSparkUILoader}
 
 object SparkDataflint {
   def install(context: SparkContext): Unit = {