Add new optimization for join

Author: Christian Herrera

docker/docker-compose.yml

@@ -40,7 +40,7 @@ services:
       - SPARK_MASTER=spark://spark-master:7077
       - SPARK_WORKER_CORES=3
       - SPARK_WORKER_MEMORY=3G
-      - SPARK_EXECUTOR_MEMORY=1G
+      - SPARK_EXECUTOR_MEMORY=3G
       - SPARK_WORKLOAD=worker
       - SPARK_LOCAL_IP=172.19.0.2
     volumes:
@@ -64,7 +64,7 @@ services:
       - SPARK_MASTER=spark://spark-master:7077
       - SPARK_WORKER_CORES=3
       - SPARK_WORKER_MEMORY=3G
-      - SPARK_EXECUTOR_MEMORY=1G
+      - SPARK_EXECUTOR_MEMORY=3G
       - SPARK_WORKLOAD=worker
       - SPARK_LOCAL_IP=172.19.0.3
     volumes:

src/main/com/codely/lesson_05_monitoring_and_optimizations/video_02_broadcast_join/JoinOptimizationApp.scala

@@ -1,10 +1,10 @@
 package com.codely.lesson_05_monitoring_and_optimizations.video_02_broadcast_join
 
-import org.apache.spark.sql.functions.{broadcast, col}
+import org.apache.spark.sql.functions.broadcast
 
 object JoinOptimizationApp extends SparkApp {
 
-  // ./bin/spark-shell --master spark://spark-master:7077 --driver-memory 4g --executor-memory 1024mb --conf spark.sql.adaptive.enabled=false
+  // ./bin/spark-shell --master spark://spark-master:7077 --driver-memory 3g --conf spark.sql.adaptive.enabled=false
 
   spark.sparkContext.setLogLevel("WARN")
 
@@ -13,22 +13,30 @@ object JoinOptimizationApp extends SparkApp {
   spark.sparkContext.setJobGroup("join without optimization", "join without optimization")
 
   val largeDF = spark
-    .range(0, 10000000L, 3) // 3.3 M
+    .range(0, 10000000L) // 10M
     .map(i => (i, s"Name$i"))
-    .toDF("id", "name")
+    .toDF("id", "fieldA")
 
-  val veryLargeDF = spark
-    .range(0, 100000000L, 2) // 50 M
+  val veryLargeDF = spark // 50 M
+    .range(0, 50000000L)
     .map(i => (i, s"Other$i"))
-    .toDF("id", "other")
+    .toDF("id", "fieldB")
 
-  veryLargeDF.join(largeDF, "id").filter(col("id") === 1).show(false)
+  veryLargeDF.join(largeDF, "id").count()
+
+  spark.sparkContext.clearJobGroup()
+
+  spark.sparkContext.setJobGroup("join with 12 shuffle partitions", "join with 12 shuffle partitions")
+
+  spark.conf.set("spark.sql.shuffle.partitions", "12")
+
+  veryLargeDF.join(largeDF, "id").count()
 
   spark.sparkContext.clearJobGroup()
 
   spark.sparkContext.setJobGroup("join with optimization", "join with optimization")
 
-  veryLargeDF.join(broadcast(largeDF), "id").show()
+  veryLargeDF.join(broadcast(largeDF), "id").count()
 
   spark.sparkContext.clearJobGroup()
 

src/main/com/codely/lesson_05_monitoring_and_optimizations/video_03_skew_join/SkewJoinApp.scala

@@ -4,53 +4,40 @@ import org.apache.spark.sql.functions.when
 
 object SkewJoinApp extends SparkApp {
 
-  // ./bin/spark-shell --master spark://spark-master:7077  --driver-memory 4g --executor-memory 1024mb --conf spark.sql.autoBroadcastJoinThreshold=-1 --conf spark.sql.adaptive.enabled=false
+  // ./bin/spark-shell --master spark://spark-master:7077  --driver-memory 3g --executor-memory 1024mb --conf spark.sql.autoBroadcastJoinThreshold=-1 --conf spark.sql.adaptive.enabled=false
 
   spark.sparkContext.setLogLevel("WARN")
 
   import spark.implicits._
 
   spark.sparkContext.setJobGroup("skewed data", "skewed data")
 
-  val skewedData = spark
-    .range(0, 10000000) // 10M
-    .withColumn("key", when($"id" < 10, $"id").otherwise(999))
-    .withColumn("value", $"id")
-
   val uniformData = spark
-    .range(0, 1000000) // 1M
+    .range(0, 10000000) // 10M
     .withColumn("key", $"id")
     .withColumn("value", $"id")
 
-  val joined = skewedData.join(uniformData, "key")
+  val skewedData = spark
+    .range(0, 200000000) // 200M
+    .withColumn("key", when($"id" < 10000000, $"id").otherwise(999))
+    .withColumn("value", $"id")
 
-  val res = joined.filter($"key" === 999).count()
-  println(s"Count for skew key (999): $res")
+  skewedData.join(uniformData, "key").count()
 
   spark.sparkContext.clearJobGroup()
 
   spark.sparkContext.setJobGroup("adaptative query execution", "adaptative query execution")
 
   spark.conf.set("spark.sql.adaptive.enabled", "true")
+  spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "1")
+  spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "20MB")
+  spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "15MB")
 
-  import org.apache.spark.sql.functions._
-
-  val skewedDataAQE = spark
-    .range(0, 10000000) // 10M
-    .withColumn("key", when($"id" < 10, $"id").otherwise(999))
-    .withColumn("value", $"id")
-
-  val uniformDataAQE = spark
-    .range(0, 1000000) // 1M
-    .withColumn("key", $"id")
-    .withColumn("value", $"id")
-
-  val joinedAQE = skewedDataAQE.join(uniformDataAQE, "key")
+  val joinedAQE = skewedData.join(uniformData, "key")
 
   joinedAQE.explain(true)
 
-  val resAQE = joinedAQE.filter($"key" === 999).count()
-  println(s"Count for skew key (999): $resAQE")
+  joinedAQE.count()
 
   spark.sparkContext.clearJobGroup()