With fair scheduler pools

snowplow-incubator · Jan 7, 2025 · 3e1534b · 3e1534b
1 parent fbd3218
commit 3e1534b
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 0 deletions.
diff --git a/modules/core/src/main/resources/fairscheduler.xml b/modules/core/src/main/resources/fairscheduler.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0"?>
+<allocations>
+  <pool name="pool1">
+    <schedulingMode>FIFO</schedulingMode>
+    <weight>1</weight>
+    <minShare>1</minShare>
+  </pool>
+  <pool name="pool2">
+    <schedulingMode>FIFO</schedulingMode>
+    <weight>1</weight>
+    <minShare>1</minShare>
+  </pool>
+</allocations>
diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/processing/SparkUtils.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/processing/SparkUtils.scala
@@ -71,6 +71,7 @@ private[processing] object SparkUtils {
     for {
       _ <- Logger[F].debug(s"Initializing local DataFrame with name $viewName")
       _ <- Sync[F].blocking {
+             spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool1")
              spark.emptyDataFrame.createTempView(viewName)
            }
     } yield ()
@@ -84,6 +85,7 @@ private[processing] object SparkUtils {
     for {
       _ <- Logger[F].debug(s"Saving batch of ${rows.size} events to local DataFrame $viewName")
       _ <- Sync[F].blocking {
+             spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool1")
              spark
                .createDataFrame(rows.toList.asJava, schema)
                .coalesce(1)
@@ -108,6 +110,7 @@ private[processing] object SparkUtils {
   def dropView[F[_]: Sync](spark: SparkSession, viewName: String): F[Unit] =
     Logger[F].info(s"Removing Spark data frame $viewName from local disk...") >>
       Sync[F].blocking {
+        spark.sparkContext.setLocalProperty("spark.scheduler.pool", "pool1")
         spark.catalog.dropTempView(viewName)
       }.void
 

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/DeltaWriter.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/DeltaWriter.scala
@@ -90,6 +90,7 @@ class DeltaWriter(config: Config.Delta) extends Writer {
     Sync[F].untilDefinedM {
       Sync[F]
         .blocking[Option[Unit]] {
+          df.sparkSession.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
           df.write
             .format("delta")
             .mode("append")

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/HudiWriter.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/HudiWriter.scala
@@ -79,6 +79,7 @@ class HudiWriter(config: Config.Hudi) extends Writer {
 
   override def write[F[_]: Sync](df: DataFrame): F[Unit] =
     Sync[F].blocking {
+      df.sparkSession.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
       df.write
         .format("hudi")
         .mode("append")

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/IcebergWriter.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/tables/IcebergWriter.scala
@@ -58,6 +58,7 @@ class IcebergWriter(config: Config.Iceberg) extends Writer {
 
   override def write[F[_]: Sync](df: DataFrame): F[Unit] =
     Sync[F].blocking {
+      df.sparkSession.sparkContext.setLocalProperty("spark.scheduler.pool", "pool2")
       df.write
         .format("iceberg")
         .mode("append")