Spark map & mapPartitions example

Author: sparkcodegeeks

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

@@ -43,9 +43,11 @@ object FlattenNestedStruct extends App {
   val df2 = df.select(col("name.*"),
     col("address.current.*"),
     col("address.previous.*"))
-  df2.toDF("fname","mename","lname","currAddState",
+
+  val df2Flatten = df2.toDF("fname","mename","lname","currAddState",
     "currAddCity","prevAddState","prevAddCity")
-    .show(false)
+  df2Flatten.printSchema()
+  df2Flatten.show(false)
 
 
 

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

@@ -0,0 +1,38 @@
+package com.sparkbyexamples.spark.dataframe.examples
+
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.types._
+
+object DataFrameComplex extends App {
+
+
+  val spark:SparkSession = SparkSession.builder()
+    .master("local[5]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  val structureData = Seq(
+    Row(Row("James","","Smith"),"36636","NewYork",3100, List("Java","Scala"),Map("hair"->"black","eye"->"brown")),
+    Row(Row("Michael","Rose",""),"40288","California",4300,List("Python","PHP"),Map("hair"->"black","eye"->"brown")),
+    Row(Row("Robert","","Williams"),"42114","Florida",1400,List("C++","C#"),Map("hair"->"black","eye"->"brown")),
+    Row(Row("Maria","Anne","Jones"),"39192","Florida",5500,List("Python","Scala"),Map("hair"->"black","eye"->"brown")),
+    Row(Row("Jen","Mary","Brown"),"34561","NewYork",3000,List("R","Scala"),Map("hair"->"black","eye"->"brown"))
+  )
+
+  val structureSchema = new StructType()
+    .add("name",new StructType()
+      .add("firstname",StringType)
+      .add("middlename",StringType)
+      .add("lastname",StringType))
+    .add("id",StringType)
+    .add("location",StringType)
+    .add("salary",IntegerType)
+    .add("languagesKnown",ArrayType(StringType))
+    .add("properties",MapType(StringType,StringType))
+
+
+  val df2 = spark.createDataFrame(spark.sparkContext.parallelize(structureData),structureSchema)
+  df2.printSchema()
+  df2.show(false)
+
+}

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

@@ -0,0 +1,58 @@
+package com.sparkbyexamples.spark.dataframe.examples
+
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.types.{IntegerType, StringType, StructType,ArrayType,MapType}
+
+object MapTransformation extends App{
+
+  val spark:SparkSession = SparkSession.builder()
+    .master("local[5]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  val structureData = Seq(
+    Row("James","","Smith","36636","NewYork",3100),
+    Row("Michael","Rose","","40288","California",4300),
+    Row("Robert","","Williams","42114","Florida",1400),
+    Row("Maria","Anne","Jones","39192","Florida",5500),
+    Row("Jen","Mary","Brown","34561","NewYork",3000)
+  )
+
+  val structureSchema = new StructType()
+    .add("firstname",StringType)
+    .add("middlename",StringType)
+    .add("lastname",StringType)
+    .add("id",StringType)
+    .add("location",StringType)
+    .add("salary",IntegerType)
+
+  val df2 = spark.createDataFrame(
+    spark.sparkContext.parallelize(structureData),structureSchema)
+  df2.printSchema()
+  df2.show(false)
+
+  import spark.implicits._
+  val util = new Util()
+  val df3 = df2.map(row=>{
+
+    val fullName = util.combine(row.getString(0),row.getString(1),row.getString(2))
+    (fullName, row.getString(3),row.getInt(5))
+  })
+  val df3Map =  df3.toDF("fullName","id","salary")
+
+  df3Map.printSchema()
+  df3Map.show(false)
+
+  val df4 = df2.mapPartitions(iterator => {
+    val util = new Util()
+    val res = iterator.map(row=>{
+      val fullName = util.combine(row.getString(0),row.getString(1),row.getString(2))
+      (fullName, row.getString(3),row.getInt(5))
+    })
+    res
+  })
+  val df4part = df4.toDF("fullName","id","salary")
+  df4part.printSchema()
+  df4part.show(false)
+
+}

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

@@ -0,0 +1,7 @@
+package com.sparkbyexamples.spark.dataframe.examples
+
+class Util extends Serializable {
+  def combine(fname:String,mname:String,lname:String):String = {
+    fname+","+mname+","+lname
+  }
+}