Spark UDF Examples

Author: sparkcodegeeks

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

@@ -4,8 +4,11 @@ import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.functions._
 
-object CastColumnType extends App{
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.functions._
 
+object CastColumnType extends App{
   val spark: SparkSession = SparkSession.builder()
     .master("local[1]")
     .appName("SparkByExamples.com")
@@ -25,26 +28,25 @@ object CastColumnType extends App{
     StructField("salary", DoubleType, true)
   ))
 
-  val df = spark.createDataFrame(spark.sparkContext.parallelize(simpleData),simpleSchema)
+  val df = spark.createDataFrame(
+    spark.sparkContext.parallelize(simpleData),simpleSchema)
   df.printSchema()
   df.show(false)
 
-  //withColumn with the original column
   val df2 = df.withColumn("age",col("age").cast(StringType))
     .withColumn("isGraduated",col("isGraduated").cast(BooleanType))
     .withColumn("jobStartDate",col("jobStartDate").cast(DateType))
   df2.printSchema()
 
-
   val df3 = df2.selectExpr("cast(age as int) age",
     "cast(isGraduated as string) isGraduated",
     "cast(jobStartDate as string) jobStartDate")
   df3.printSchema()
   df3.show(false)
 
   df3.createOrReplaceTempView("CastExample")
-  val df4 = spark.sql("SELECT STRING(age),BOOLEAN(isGraduated),DATE(jobStartDate) from CastExample")
-  df4.printSchema()
-  df4.show(false)
-
-}
+  val df4 = spark.sql("SELECT STRING(age),BOOLEAN(isGraduated), " +
+    "DATE(jobStartDate) from CastExample")
+    df4.printSchema()
+    df4.show(false)
+    }

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

@@ -13,6 +13,10 @@ object FromCSVFile2 {
 
     val filePath="src/main/resources/stream.csv"
 
+    val df3 = spark.read.option("header",true).csv("src/main/resources/zipcodes.csv")
+    df3.show(false)
+
+
     val df = spark.read.options(Map("inferSchema"->"true","delimiter"->"|","header"->"true")).csv(filePath)
 
     val df2 = df.select("Gender", "BirthDate", "TotalCost", "TotalChildren", "ProductCategoryName")

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

@@ -0,0 +1,39 @@
+package com.sparkbyexamples.spark.dataframe
+
+import org.apache.spark.sql.functions.udf
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.{Row, SparkSession}
+
+object SparkUDF extends App{
+
+  val spark: SparkSession = SparkSession.builder()
+    .master("local[1]")
+    .appName("SparkByExamples.com")
+    .getOrCreate()
+
+  import spark.implicits._
+  val columns = Seq("Seqno","Quote")
+  val data = Seq(("1", "Be the change that you wish to see in the world"),
+    ("2", "Everyone thinks of changing the world, but no one thinks of changing himself."),
+    ("3", "The purpose of our lives is to be happy.")
+
+  )
+  val df = data.toDF(columns:_*)
+  df.show(false)
+
+  val convertCase =  (str:String) => {
+    val arr = str.split(" ")
+    arr.map(f=>  f.substring(0,1).toUpperCase + f.substring(1,f.length)).mkString(" ")
+  }
+
+  //Using with DataFrame
+  val convertUDF = udf(convertCase)
+  df.select(col("Seqno"),
+    convertUDF(col("Quote")).as("Quote") ).show(false)
+
+  // Using it on SQL
+  spark.udf.register("convertUDF", convertCase)
+  df.createOrReplaceTempView("QUOTE_TABLE")
+  spark.sql("select Seqno, convertUDF(Quote) from QUOTE_TABLE").show(false)
+
+}

src/main/scala/com/sparkbyexamples/spark/dataframe/functions/collection/MapToColumn.scala

@@ -16,21 +16,21 @@ object DateInMilli extends App{
 
   val df = Seq(1).toDF("seq").select(
     current_date().as("current_date"),
-    unix_timestamp().as("milliseconds")
-    )
+    unix_timestamp().as("unix_timestamp_seconds")
+  )
 
   df.printSchema()
   df.show(false)
 
-  //Convert milliseconds to date
+  //Convert unix seconds to date
   df.select(
-    to_date(col("milliseconds").cast(TimestampType)).as("current_date")
+    to_date(col("unix_timestamp_seconds").cast(TimestampType)).as("current_date")
   ).show(false)
 
-  //convert date to milliseconds
+  //convert date to unix seconds
   df.select(
-    unix_timestamp(col("current_date")).as("unix_milliseconds"),
-    unix_timestamp(lit("12-21-2019"),"mm-DD-yyyy").as("unix_milliseconds2")
+    unix_timestamp(col("current_date")).as("unix_seconds"),
+    unix_timestamp(lit("12-21-2019"),"mm-DD-yyyy").as("unix_seconds2")
   ).show(false)
 
-}
+}

src/main/scala/com/sparkbyexamples/spark/dataframe/functions/datetime/DateInMilli.scala

@@ -16,22 +16,22 @@ object TimeInMilli extends App{
 
   val df = Seq(1).toDF("seq").select(
     current_timestamp().as("current_time"),
-    unix_timestamp().as("milliseconds")
-    )
+    unix_timestamp().as("epoch_time_seconds")
+  )
 
   df.printSchema()
   df.show(false)
 
-  //Convert milliseconds to timestamp
+  //Convert epoch_time to timestamp
   df.select(
-    col("milliseconds").cast(TimestampType).as("current_time"),
-      col("milliseconds").cast("timestamp").as("current_time2")
+    col("epoch_time_seconds").cast(TimestampType).as("current_time"),
+    col("epoch_time_seconds").cast("timestamp").as("current_time2")
   ).show(false)
 
-  //convert timestamp to milliseconds
+  //convert timestamp to Unix epoch time
   df.select(
-    unix_timestamp(col("current_time")).as("unix_milliseconds"),
-    col("current_time").cast(LongType).as("time_to_milli")
+    unix_timestamp(col("current_time")).as("unix_epoch_time"),
+    col("current_time").cast(LongType).as("unix_epoch_time2")
   ).show(false)
 
 }