spark examples

Author: sparkcodegeeks

src/main/resources/address.csv

@@ -0,0 +1,6 @@
+Id,Address Line1,City,State,Zipcode
+1,9182 Clear Water Rd,Fayetteville,AR,72704
+2,9724 E Landon Ln,Kennewick,WA,99338
+3,9509 Clay Creek Ln,Fort Worth,TX,76177
+4,98016 S Garnsey St,Santa Ana,CA,92707
+5,9920 State Highway 89,Ringling,OK,73456

src/main/scala/com/sparkbyexamples/spark/dataframe/CreateDataFrame.scala

@@ -1,15 +1,14 @@
 package com.sparkbyexamples.spark.dataframe
 
 import org.apache.spark.sql.types.{StringType, StructField, StructType}
-import org.apache.spark.sql.{DataFrame, Row, SparkSession}
+import org.apache.spark.sql.{Row, SparkSession}
 
 object CreateDataFrame {
 
   def main(args:Array[String]):Unit={
 
     val spark:SparkSession = SparkSession.builder()
-      .master("local[1]")
-      .appName("SparkByExample")
+      .master("local[1]").appName("SparkByExamples.com")
       .getOrCreate()
 
     import spark.implicits._
@@ -20,10 +19,10 @@ object CreateDataFrame {
 
     //From RDD (USING toDF())
     val dfFromRDD1 = rdd.toDF("language","users")
-
+    dfFromRDD1.printSchema()
     //From RDD (USING createDataFrame)
     val dfFromRDD2 = spark.createDataFrame(rdd).toDF(columns:_*)
-
+    dfFromRDD2.printSchema()
     //From RDD (USING createDataFrame and Adding schema using StructType)
     //convert RDD[T] to RDD[Row]
     val schema = StructType( Array(StructField("language", StringType,  true),

src/main/scala/com/sparkbyexamples/spark/dataframe/CreateEmptyDatasetExample.scala

@@ -10,7 +10,7 @@ object CreateEmptyDatasetExample extends App {
     .master("local[1]")
     .appName("SparkByExamples.com")
     .getOrCreate()
-
+  spark.sparkContext.setLogLevel("ERROR");
   import spark.implicits._
 
   val schema = StructType(
@@ -21,11 +21,24 @@ object CreateEmptyDatasetExample extends App {
   val colSeq = Seq("firstName","lastName","middleName")
 
   case class Name(firstName: String, lastName: String, middleName:String)
+  case class Empty()
+  val ds0 = spark.emptyDataset[Empty]
+  ds0.printSchema()
+
+  val ds1=spark.emptyDataset[Name]
+  ds1.printSchema()
+
+  val ds2 = spark.createDataset(Seq.empty[Name])
+  ds2.printSchema()
+
+  val ds4=spark.createDataset(spark.sparkContext.emptyRDD[Name])
+  ds4.printSchema()
+
+  val ds3=spark.createDataset(Seq.empty[(String,String,String)])
+  ds3.printSchema()
+  val ds5=Seq.empty[(String,String,String)].toDS()
+  ds5.printSchema()
 
-  spark.createDataset(Seq.empty[Name])
-  spark.createDataset(Seq.empty[(String,String,String)])
-  spark.createDataset(spark.sparkContext.emptyRDD[Name])
-  Seq.empty[(String,String,String)].toDS()
-  Seq.empty[Name].toDS()
-  spark.emptyDataset[Name]
+  val ds6=Seq.empty[Name].toDS()
+  ds6.printSchema()
 }

src/main/scala/com/sparkbyexamples/spark/dataframe/FilterNullRowsExample.scala

@@ -0,0 +1,43 @@
+package com.sparkbyexamples.spark.dataframe
+
+import org.apache.spark.sql.{SparkSession}
+import org.apache.spark.sql.functions.col
+
+object FilterNullRowsExample extends App{
+
+  val spark: SparkSession = SparkSession.builder()
+    .master("local[1]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  spark.sparkContext.setLogLevel("ERROR")
+   val data = Seq(
+    ("James",null,"M"),
+    ("Anna","NY","F"),
+    ("Julia",null,null)
+  )
+  import spark.implicits._
+  val columns = Seq("name","state","gender")
+  val df = data.toDF(columns:_*)
+
+  df.printSchema()
+  df.show()
+
+  df.filter("state is NULL").show(false)
+  df.filter(df("state").isNull).show(false)
+  df.filter(col("state").isNull).show(false)
+
+  df.filter("state is not NULL").show(false)
+  df.filter("NOT state is NULL").show(false)
+  df.filter(df("state").isNotNull).show(false)
+
+  df.filter("state is NULL AND gender is NULL").show(false)
+  df.filter(df("state").isNull && df("gender").isNull).show(false)
+
+  df.createOrReplaceTempView("DATA")
+  spark.sql("SELECT * FROM DATA where STATE IS NULL").show(false)
+  spark.sql("SELECT * FROM DATA where STATE IS NULL AND GENDER IS NULL").show(false)
+  spark.sql("SELECT * FROM DATA where STATE IS NOT NULL").show(false)
+
+
+}

src/main/scala/com/sparkbyexamples/spark/dataframe/RemoveNullRowsExample.scala

@@ -9,14 +9,18 @@ object RemoveNullRowsExample extends App{
     .appName("SparkByExamples.com")
     .getOrCreate()
 
+  spark.sparkContext.setLogLevel("ERROR")
   val filePath="src/main/resources/small_zipcode.csv"
 
   val df = spark.read.options(Map("inferSchema"->"true","delimiter"->",","header"->"true")).csv(filePath)
   df.printSchema()
   df.show(false)
 
-  df.na.drop()
-    .show(false)
+  df.na.drop().show(false)
+
+  //all/any
+  df.na.drop("any").show(false)
+
+  df.na.drop(Seq("population","type")).show(false)
 
-  // Array and map columns
 }

src/main/scala/com/sparkbyexamples/spark/dataframe/WithColumn.scala

@@ -1,7 +1,7 @@
 package com.sparkbyexamples.spark.dataframe
 
 import org.apache.spark.sql.{Row, SparkSession}
-import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType, StringType, StructType}
+import org.apache.spark.sql.types.{StringType, StructType}
 import org.apache.spark.sql.functions._
 object WithColumn {
 
@@ -12,27 +12,23 @@ object WithColumn {
       .appName("SparkByExamples.com")
       .getOrCreate()
 
-    val arrayStructureData = Seq(
-      Row(Row("James ","","Smith"),"1","M",3100,List("Cricket","Movies"),Map("hair"->"black","eye"->"brown")),
-      Row(Row("Michael ","Rose",""),"2","M",3100,List("Tennis"),Map("hair"->"brown","eye"->"black")),
-      Row(Row("Robert ","","Williams"),"3","M",3100,List("Cooking","Football"),Map("hair"->"red","eye"->"gray")),
-      Row(Row("Maria ","Anne","Jones"),"4","M",3100,null,Map("hair"->"blond","eye"->"red")),
-      Row(Row("Jen","Mary","Brown"),"5","M",3100,List("Blogging"),Map("white"->"black","eye"->"black"))
+    val dataRows = Seq(Row(Row("James;","","Smith"),"36636","M","3000"),
+      Row(Row("Michael","Rose",""),"40288","M","4000"),
+      Row(Row("Robert","","Williams"),"42114","M","4000"),
+      Row(Row("Maria","Anne","Jones"),"39192","F","4000"),
+      Row(Row("Jen","Mary","Brown"),"","F","-1")
     )
 
-    val arrayStructureSchema = new StructType()
+    val schema = new StructType()
       .add("name",new StructType()
         .add("firstname",StringType)
         .add("middlename",StringType)
         .add("lastname",StringType))
-      .add("id",StringType)
+      .add("dob",StringType)
       .add("gender",StringType)
-      .add("salary",IntegerType)
-      .add("Hobbies", ArrayType(StringType))
-      .add("properties", MapType(StringType,StringType))
+      .add("salary",StringType)
 
-    val df2 = spark.createDataFrame(
-      spark.sparkContext.parallelize(arrayStructureData),arrayStructureSchema)
+    val df2 = spark.createDataFrame(spark.sparkContext.parallelize(dataRows),schema)
 
     //Change the column data type
     df2.withColumn("salary",df2("salary").cast("Integer"))
@@ -62,10 +58,6 @@ object WithColumn {
     df2.select("name.firstname").show(false)
     df2.select("name.*").show(false)
 
-
-    val df8 = df2.select(col("*"),explode(col("hobbies")))
-    df8.show(false)
-
     import spark.implicits._
 
     val columns = Seq("name","address")
@@ -81,5 +73,9 @@ object WithColumn {
     val finalDF = newDF.toDF("First Name","Last Name","Address Line1","City","State","zipCode")
     finalDF.printSchema()
     finalDF.show(false)
+
+    df2.createOrReplaceTempView("PERSON")
+    spark.sql("SELECT salary*100 as salary, salary*-1 as CopiedColumn, 'USA' as country FROM PERSON").show()
   }
+
 }

src/main/scala/com/sparkbyexamples/spark/dataframe/examples/ForEachPartExample.scala

@@ -1,33 +1,33 @@
-package com.sparkbyexamples.spark.dataframe.examples
-
-import org.apache.spark.sql.SparkSession
-
-object ForEachPartExample extends App {
-
-  val spark: SparkSession = SparkSession.builder()
-    .master("local[1]")
-    .appName("SparkByExamples.com")
-    .getOrCreate()
-
-  val data = Seq(("Banana",1000,"USA"), ("Carrots",1500,"USA"), ("Beans",1600,"USA"),
-    ("Orange",2000,"USA"),("Orange",2000,"USA"),("Banana",400,"China"),
-    ("Carrots",1200,"China"),("Beans",1500,"China"))
-
-  // foreachPartition DataFrame
-  val df = spark.createDataFrame(data).toDF("Product","Amount","Country")
-  df.foreachPartition(partition => {
-    //Initialize any database connection
-    partition.foreach(fun=>{
-      //apply the function
-    })
-  })
-
-  //rdd
-  val rdd = spark.sparkContext.parallelize(Seq(1,2,3,4,5,6,7,8,9))
-  rdd.foreachPartition(partition => {
-    //Initialize any database connection
-    partition.foreach(fun=>{
-      //apply the function
-    })
-  })
-}
+//package com.sparkbyexamples.spark.dataframe.examples
+//
+//import org.apache.spark.sql.SparkSession
+//
+//object ForEachPartExample extends App {
+//
+//  val spark: SparkSession = SparkSession.builder()
+//    .master("local[1]")
+//    .appName("SparkByExamples.com")
+//    .getOrCreate()
+//
+//  val data = Seq(("Banana",1000,"USA"), ("Carrots",1500,"USA"), ("Beans",1600,"USA"),
+//    ("Orange",2000,"USA"),("Orange",2000,"USA"),("Banana",400,"China"),
+//    ("Carrots",1200,"China"),("Beans",1500,"China"))
+//
+//  // foreachPartition DataFrame
+//  val df = spark.createDataFrame(data).toDF("Product","Amount","Country")
+//  df.foreachPartition(partition => {
+//    //Initialize any database connection
+//    partition.foreach(fun=>{
+//      //apply the function
+//    })
+//  })
+//
+//  //rdd
+//  val rdd = spark.sparkContext.parallelize(Seq(1,2,3,4,5,6,7,8,9))
+//  rdd.foreachPartition(partition => {
+//    //Initialize any database connection
+//    partition.foreach(fun=>{
+//      //apply the function
+//    })
+//  })
+//}

src/main/scala/com/sparkbyexamples/spark/dataframe/examples/SaveSingleFile.scala

@@ -0,0 +1,40 @@
+package com.sparkbyexamples.spark.dataframe.examples
+
+import java.io.File
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
+import org.apache.spark.sql.{SaveMode, SparkSession}
+
+object SaveSingleFile extends App{
+
+  val spark:SparkSession = SparkSession.builder()
+    .master("local[3]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  val df = spark.read.option("header",true).csv("src/main/resources/address.csv")
+  df.repartition(1).write.mode(SaveMode.Overwrite).csv("/tmp/address")
+
+
+  val hadoopConfig = new Configuration()
+  val hdfs = FileSystem.get(hadoopConfig)
+
+  val srcPath=new Path("/tmp/address")
+  val destPath= new Path("/tmp/address_merged.csv")
+  val srcFile=FileUtil.listFiles(new File("c:/tmp/address")).filterNot(f=>f.getPath.endsWith(".csv"))(0)
+  //Copy the CSV file outside of Directory and rename
+  FileUtil.copy(srcFile,hdfs,destPath,true,hadoopConfig)
+  //Remove Directory created by df.write()
+  hdfs.delete(srcPath,true)
+  //Removes CRC File
+  hdfs.delete(new Path("/tmp/.address_merged.csv.crc"),true)
+
+  // Merge Using Haddop API
+  df.repartition(1).write.mode(SaveMode.Overwrite).csv("/tmp/address-tmp")
+  val srcFilePath=new Path("/tmp/address-tmp")
+  val destFilePath= new Path("/tmp/address_merged2.csv")
+  FileUtil.copyMerge(hdfs, srcFilePath, hdfs, destFilePath, true, hadoopConfig, null)
+  //Remove hidden CRC file if not needed.
+  hdfs.delete(new Path("/tmp/.address_merged2.csv.crc"),true)
+}

src/main/scala/com/sparkbyexamples/spark/rdd/functions/SortByKeyExample.scala

@@ -0,0 +1,28 @@
+package com.sparkbyexamples.spark.rdd.functions
+
+import org.apache.spark.sql.SparkSession
+
+object SortByKeyExample extends App{
+
+  val spark: SparkSession = SparkSession.builder()
+    .master("local[1]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  val data = Seq(("Project","A", 1),
+    ("Gutenberg’s", "X",3),
+    ("Alice’s", "C",5),
+    ("Adventures","B", 1)
+  )
+
+  val rdd=spark.sparkContext.parallelize(data)
+  rdd.foreach(println)
+  val rdd2=rdd.map(f=>{(f._2, (f._1,f._2,f._3))})
+  rdd2.foreach(println)
+  val rdd3= rdd2.sortByKey()
+  val rdd4= rdd2.sortByKey(false)
+  rdd4.foreach(println)
+
+  val rdd5 = rdd.sortBy(f=>(f._3,f._2),false)
+  rdd5.foreach(println)
+}