Introduction to RDD, Dataset and Dataframe

satendrakumar · satendrakumar · commit fa1337d506f3 · 2021-07-27T20:10:55.000+05:30
diff --git a/.gitignore b/.gitignore
@@ -0,0 +1,9 @@
+logs
+target
+/.idea
+/.idea_modules
+/.classpath
+/.project
+/.settings
+/RUNNING_PID
+*.log
diff --git a/README.md b/README.md
@@ -0,0 +1,17 @@
+# learning-spark
+
+##### Run dataset operation 
+```shell script
+ sbt "runMain com.techmonad.learn.DataSetOps"
+```
+
+##### Run dataframe operation 
+```shell script
+ sbt "runMain com.techmonad.learn.DataFrameOps"
+```
+
+
+##### Run RDD operation 
+```shell script
+ sbt "runMain com.techmonad.learn.RDDOps"
+```
diff --git a/build.sbt b/build.sbt
@@ -0,0 +1,10 @@
+name := "learning-spark"
+
+version := "0.1"
+
+scalaVersion := "2.12.10"
+
+
+libraryDependencies ++= Seq(
+  "org.apache.spark" %% "spark-sql" % "3.1.2"
+)
diff --git a/data/tweets.csv b/data/tweets.csv
diff --git a/data/user-details.csv b/data/user-details.csv
@@ -0,0 +1,6 @@
+id,employer,location,salary
+1,  Google, SFO,2000
+2, Yahoo, LA,5000
+3, Microsoft, GA,10000
+10, Apple, CA,9000
+1,  Google, SFO,120000
diff --git a/data/users.csv b/data/users.csv
@@ -0,0 +1,5 @@
+id,name,email
+1, Bob, bob@gmail.com
+2, Joy, joy@gmail.com
+3, Rob, rob@gmail.com
+4, Blah, blah@gmail.com
diff --git a/data/words.txt b/data/words.txt
@@ -0,0 +1,6 @@
+bob rob rob
+bob joy jai
+rob
+bob
+joy
+jai
diff --git a/project/build.properties b/project/build.properties
@@ -0,0 +1 @@
+sbt.version = 1.3.8
diff --git a/src/main/resources/log4j.properties b/src/main/resources/log4j.properties
@@ -0,0 +1,12 @@
+PATTERN =[%-5p]-[%d]-[%c]-[%m]%n
+
+log4j.rootLogger=WARN, consoleAppender, fileAppender
+
+log4j.appender.consoleAppender=org.apache.log4j.ConsoleAppender
+log4j.appender.consoleAppender.layout=org.apache.log4j.PatternLayout
+log4j.appender.consoleAppender.layout.ConversionPattern=${PATTERN}
+
+log4j.appender.fileAppender=org.apache.log4j.RollingFileAppender
+log4j.appender.fileAppender.layout=org.apache.log4j.PatternLayout
+log4j.appender.fileAppender.layout.ConversionPattern=${PATTERN}
+log4j.appender.fileAppender.File=sparkApp.log
diff --git a/src/main/scala/com/techmonad/learn/DataFrameOps.scala b/src/main/scala/com/techmonad/learn/DataFrameOps.scala
@@ -0,0 +1,62 @@
+package com.techmonad.learn
+
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.functions._
+
+object DataFrameOps extends SparkSessionProvider {
+
+  def main(args: Array[String]): Unit = {
+
+    val df: DataFrame =
+      spark
+        .read
+        .text("data/words.txt")
+    println("###############Word Count################")
+    val wordCounts =
+      df
+        .withColumn("words", split(col("value"), "\\s+"))
+        .withColumn("word", explode(col("words")))
+        .select("word")
+        .groupBy("word")
+        .agg(count("word").as("count"))
+    wordCounts.show()
+
+    // Joins
+    val users =
+      spark
+        .read
+        .option("delimiter", ",")
+        .option("header", "true")
+        .csv("data/users.csv")
+    users.show()
+
+    val userDetails =
+      spark
+        .read
+        .option("header", "true")
+        .option("delimiter", ",")
+        .csv("data/user-details.csv")
+
+    userDetails.show()
+
+    println("#############Inner Join###################")
+    val innerJoin: DataFrame = users.join(userDetails, Seq("id"), "inner")
+    innerJoin.show()
+
+    println("#############Left Join###################")
+    val leftJoin: DataFrame = users.join(userDetails, Seq("id"), "left")
+    leftJoin.show()
+
+    println("#############Right Join###################")
+    val rightJoin: DataFrame = users.join(userDetails, Seq("id"), "right")
+    rightJoin.show()
+
+
+    println("#############full Join###################")
+    val fullJoin: DataFrame = users.join(userDetails, Seq("id"), "full")
+    fullJoin.show()
+
+    spark.stop()
+  }
+
+}
diff --git a/src/main/scala/com/techmonad/learn/DataSetOps.scala b/src/main/scala/com/techmonad/learn/DataSetOps.scala
@@ -0,0 +1,99 @@
+package com.techmonad.learn
+
+import org.apache.spark.sql.functions._
+import org.apache.spark.sql.{Dataset, Encoders}
+
+
+object DataSetOps extends SparkSessionProvider {
+
+  def main(args: Array[String]): Unit = {
+
+    import spark.implicits._
+
+    // read text file as a dataset
+    val ds: Dataset[String] =
+      spark
+        .read
+        .textFile("data/words.txt")
+
+    println("############# Word Count ###################")
+    val wordCounts: Dataset[(String, Long)] =
+      ds
+        .flatMap { line =>
+          line
+            .split("\\s+")
+            .map(_.trim)
+        }
+        .filter { word => word.length > 0 }
+        .groupByKey { word => word }
+        .count()
+
+    wordCounts.show()
+
+    // read csv file as a dataset
+    val users: Dataset[User] =
+      spark
+        .read
+        .schema(Encoders.product[User].schema)
+        .option("delimiter", ",")
+        .option("header", "true")
+        .csv("data/users.csv")
+        .as[User]
+    users.show()
+
+    val userDetails: Dataset[Detail] =
+      spark
+        .read
+        .schema(Encoders.product[Detail].schema)
+        .option("header", "true")
+        .option("delimiter", ",")
+        .csv("data/user-details.csv")
+        .as[Detail]
+
+    userDetails.show()
+
+    /**
+     * Joins - inner, left, right, ful
+     **/
+
+    // inner join(by default is inner join)
+    // val innerJoin = users.join(userDetails, "id")
+    // more explicit
+    println("#############Inner Join###################")
+    val innerJoin: Dataset[UserDetails] = users.join(userDetails, Seq("id"), "inner").as[UserDetails]
+    innerJoin.show()
+
+    println("#############Left Join###################")
+    val leftJoin: Dataset[UserDetails] = users.join(userDetails, Seq("id"), "left").as[UserDetails]
+    leftJoin.show()
+
+    println("#############Right Join###################")
+    val rightJoin: Dataset[UserDetails] = users.join(userDetails, Seq("id"), "right").as[UserDetails]
+    rightJoin.show()
+
+
+    println("#############full Join###################")
+    val fullJoin: Dataset[UserDetails] = users.join(userDetails, Seq("id"), "full").as[UserDetails]
+    fullJoin.show()
+
+    println("#############JoinWith ###################")
+    val usingJoinWith: Dataset[(User, Detail)] = users.joinWith(userDetails, users("id") === userDetails("id"))
+    usingJoinWith.show()
+
+    // Agg
+    // Get max, avg, min  salary
+    println("############# Max  salary###################")
+    innerJoin
+      .groupBy($"name")
+      .agg(max("salary").as("max salary"), avg("salary").as("Average salary"), min("salary").as("min salary"))
+      .show()
+
+    spark.stop()
+  }
+}
+
+case class User(id: Int, name: String, email: String)
+
+case class Detail(id: Int, employer: String, location: String, salary: Int)
+
+case class UserDetails(id: Option[Int], name: Option[String], email: Option[String], employer: Option[String], location: Option[String], salary: Option[Int])
diff --git a/src/main/scala/com/techmonad/learn/RDDOps.scala b/src/main/scala/com/techmonad/learn/RDDOps.scala
@@ -0,0 +1,74 @@
+package com.techmonad.learn
+
+import org.apache.spark.rdd.RDD
+
+object RDDOps extends SparkSessionProvider {
+
+  def main(args: Array[String]): Unit = {
+
+    // text file reading
+    val rdd: RDD[String] =
+      sc
+        .textFile("data/words.txt")
+
+    println("############ Word count ##############################")
+    val wordCounts: RDD[(String, Int)] =
+      rdd
+        .flatMap { line =>
+          line
+            .split("\\s+")
+            .filter { word => word.length > 0 }
+        }
+        .map { word => (word, 1) }
+        .reduceByKey { case (count1, count2) => count1 + count2 }
+
+    wordCounts.collect.foreach(println)
+
+
+    val users: RDD[User] =
+      sc
+        .textFile("data/users.csv")
+        .mapPartitionsWithIndex { case (index, itr) => if (index == 0) itr.drop(1) else itr }
+        .map { line =>
+          val arr =
+            line
+              .split(",")
+              .map(_.trim)
+          User(arr(0).toInt, arr(1), arr(2))
+        }
+
+    users.collect.foreach(println)
+
+    val details: RDD[Detail] =
+      sc
+        .textFile("data/user-details.csv")
+        .mapPartitionsWithIndex { case (index, itr) => if (index == 0) itr.drop(1) else itr }
+        .map { line =>
+          val arr =
+            line
+              .split(",")
+              .map(_.trim)
+          Detail(arr(0).toInt, arr(1), arr(2), arr(3).toInt)
+        }
+
+    details.collect.foreach(println)
+
+
+    val userWithId: RDD[(Int, User)] =users.map { user => (user.id, user) }
+    val detailWithId: RDD[(Int, Detail)] = details.map{ detail => (detail.id, detail) }
+    // Joins in RDD
+    val userDetails: RDD[(Int, (User, Detail))] = userWithId.join(detailWithId)
+
+     userDetails.collect.foreach(println)
+
+    val userDetailsLeft: RDD[(Int, (User, Option[Detail]))] = userWithId.leftOuterJoin(detailWithId)
+    userDetailsLeft.collect.foreach(println)
+
+    val userDetailsRight: RDD[(Int, (Option[User], Detail))] = userWithId.rightOuterJoin(detailWithId)
+    userDetailsRight.collect.foreach(println)
+
+    spark.stop()
+  }
+
+
+}
diff --git a/src/main/scala/com/techmonad/learn/SparkSessionProvider.scala b/src/main/scala/com/techmonad/learn/SparkSessionProvider.scala
@@ -0,0 +1,17 @@
+package com.techmonad.learn
+
+import org.apache.spark.sql.SparkSession
+
+trait SparkSessionProvider {
+
+  val spark =
+    SparkSession
+      .builder()
+      .appName("LearningSpark")
+      .master("local[*]")
+      .getOrCreate()
+
+  val sc = spark.sparkContext
+
+
+}

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +id,name,email
 +1, Bob, [email protected]
 +2, Joy, [email protected]
 +3, Rob, [email protected]
 +4, Blah, [email protected]