DF level 2

Author: satendrakumar

data/emp.csv

@@ -0,0 +1,19 @@
+empid,ename,salary,deptId,mgtId
+1, ram,7,1,3
+2, shan,3,2,3
+3, rohit,50,2,null
+4,mohit,10,3,2
+5, ss,53,3,2
+6,eeee,56,3,1
+7,ddd,32,4,1
+8,ppp,35,4,4
+9,lll,5,4,4
+10,tt,43,5,5
+11,ccc,71,5,5
+12,ggg,72,5,5
+13,ccc,73,5,5
+14,ggg,74,5,5
+15,ddd,82,4,1
+16,ppp,55,4,4
+17,lll,59,4,4
+

src/main/scala/com/techmonad/learn/DataFrameOps.scala

@@ -19,7 +19,7 @@ object DataFrameOps extends SparkSessionProvider {
         .select("word")
         .groupBy("word")
         .agg(count("word").as("count"))
-    wordCounts.show()
+    // wordCounts.show()
 
     // Joins
     val users =
@@ -28,7 +28,7 @@ object DataFrameOps extends SparkSessionProvider {
         .option("delimiter", ",")
         .option("header", "true")
         .csv("data/users.csv")
-    users.show()
+    //users.show()
 
     val userDetails =
       spark
@@ -37,26 +37,31 @@ object DataFrameOps extends SparkSessionProvider {
         .option("delimiter", ",")
         .csv("data/user-details.csv")
 
-    userDetails.show()
+    //userDetails.show()
 
     println("#############Inner Join###################")
     val innerJoin: DataFrame = users.join(userDetails, Seq("id"), "inner")
-    innerJoin.show()
+    //innerJoin.show()
 
     println("#############Left Join###################")
     val leftJoin: DataFrame = users.join(userDetails, Seq("id"), "left")
-    leftJoin.show()
+    //leftJoin.show()
 
     println("#############Right Join###################")
     val rightJoin: DataFrame = users.join(userDetails, Seq("id"), "right")
-    rightJoin.show()
+    //rightJoin.show()
 
 
     println("#############full Join###################")
     val fullJoin: DataFrame = users.join(userDetails, Seq("id"), "full")
     fullJoin.show()
 
+
     spark.stop()
   }
 
 }
+
+
+
+

src/main/scala/com/techmonad/learn/DataFrameOpsLevel2.scala

@@ -0,0 +1,60 @@
+package com.techmonad.learn
+
+
+import org.apache.spark.sql.expressions.{Window, WindowSpec}
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+
+
+object DataFrameOpsLevel2 extends App with SparkSessionProvider {
+
+  // top # elements by department
+  //CSV => empid,ename,salary,deptId,mgtId
+
+  val schema =
+    StructType(
+      Seq(
+        StructField("empid", IntegerType, false),
+        StructField("ename", StringType, false),
+        StructField("salary", IntegerType, false),
+        StructField("deptId", IntegerType, false),
+        StructField("mgtId", IntegerType, true)
+      )
+    )
+
+  val empDF =
+    spark
+      .read
+      .schema(schema)
+      .option("header", "true")
+      .option("sep", ",")
+      .csv("data/emp.csv")
+
+
+  val partitionByDeptOrderBySal: WindowSpec =
+    Window
+      .partitionBy("deptId")
+      .orderBy(desc("salary"))
+
+
+  // TOP 5 salary in each department
+  empDF
+    .withColumn("rank", dense_rank() over partitionByDeptOrderBySal)
+    .filter("rank <= 5")
+    .select("deptId", "salary")
+    .show(20)
+
+
+  /** Expensive apprach
+   *spark.udf.register("sort_list",(list:List[Int]) =>   list.sorted.take(3))
+   * *
+   * empDF
+   * .groupBy("deptId")
+   * .agg(collect_list($"salary").as("list"))
+   * .selectExpr("deptId", "sort_list(list)")
+   * .show()
+   *
+   */
+
+
+}

src/main/scala/com/techmonad/learn/RDDOps.scala

@@ -2,6 +2,7 @@ package com.techmonad.learn
 
 import org.apache.spark.broadcast.Broadcast
 import org.apache.spark.rdd.RDD
+import org.apache.spark.storage.StorageLevel
 import org.apache.spark.util.LongAccumulator
 
 object RDDOps extends SparkSessionProvider {
@@ -13,6 +14,9 @@ object RDDOps extends SparkSessionProvider {
       sc
         .textFile("data/words.txt")
 
+    // persist RDD in memory
+     rdd.persist(StorageLevel.MEMORY_ONLY)
+
     println("############ Word count ##############################")
     val wordCounts: RDD[(String, Int)] =
       rdd

src/main/scala/com/techmonad/learn/StructuredStreamingOps.scala

@@ -34,6 +34,7 @@ object StructuredStreamingOps extends SparkSessionProvider {
 
     query.awaitTermination()
 
+
     /*    words
           .writeStream
           .foreachBatch { (df, batchNo) =>