added UDF

Author: mahmoudparsian

code/bonus_chapters/README.md

@@ -14,7 +14,7 @@ so I added the following bonus chapters online.
 | [K-mers](./k-mers/)                         | K-mers for DNA Sequences |
 | [Correlation](./correlation/)               | All vs. All Correlation |
 | [`mapPartitions()` Transformation](./mappartitions/) | `mapPartitions()` Complete Example  |
-
+| [`UDF`](./UDF/) | User-Defined Function Example  |
 
 -----
 

code/bonus_chapters/UDF/python/README.md

@@ -0,0 +1 @@
+Demo Spark's UDF (user-defined-function)

code/bonus_chapters/UDF/python/dataframe_UDF_example.log

@@ -0,0 +1,39 @@
+% export SPARK_HOME=/home/mparsian/spark-3.2.0
+% $SPARK_HOME/bin/spark-submit dataframe_UDF_example.py
+
++---+------------+
+|ID |Name        |
++---+------------+
+|100|john jones  |
+|200|tracey smith|
+|300|amy sanders |
+|400|null        |
++---+------------+
+
++---+------------+
+|ID |Name        |
++---+------------+
+|100|John Jones  |
+|200|Tracey Smith|
+|300|Amy Sanders |
+|400|null        |
++---+------------+
+
++---+------------+------------+
+|ID |Name        |Upper Name  |
++---+------------+------------+
+|100|john jones  |JOHN JONES  |
+|200|tracey smith|TRACEY SMITH|
+|300|amy sanders |AMY SANDERS |
+|400|null        |null        |
++---+------------+------------+
+
++---+------------+
+|ID |Name        |
++---+------------+
+|100|John Jones  |
+|200|Tracey Smith|
+|300|Amy Sanders |
+|400|null        |
++---+------------+
+

code/bonus_chapters/UDF/python/dataframe_UDF_example.py

@@ -0,0 +1,65 @@
+# import required libraries
+from pyspark.sql import SparkSession
+from pyspark.sql.functions import col
+from pyspark.sql.functions import udf
+from pyspark.sql.types import StringType
+
+#--------------------------------------------------
+# Demo concept of Spark UDF (user-defined-function)
+#--------------------------------------------------
+# @author: Mahmoud Parsian
+#--------------------------------------------------
+def convert_case(name):
+    if name is None: return None
+    if len(name) < 1: return ""
+    result_string = ""
+    arr = name.split(" ")
+    for x in arr:
+       result_string += x[0:1].upper() + x[1:len(x)] + " "
+    #end-for
+    return result_string.strip()
+#end-def
+#--------------------------------------------------
+def to_upper_case(name):
+    if name is None: return None
+    if len(name) < 1: return ""    
+    return name.upper()
+#end-def
+#--------------------------------------------------
+#
+# create a SparkSession object
+spark = SparkSession.builder.appName('UDF-Learning').getOrCreate()
+
+# define column names for a DataFrame
+column_names = ["ID", "Name"]
+
+# define some rows for a DataFrame
+some_data = [("100", "john jones"),
+             ("200", "tracey smith"),
+             ("300", "amy sanders"),
+             ("400", None)]
+
+# create a DataFrame
+df = spark.createDataFrame(data=some_data,schema=column_names)
+
+# display content of a DataFrame for testing/debugging
+df.show(truncate=False)
+
+
+# Converting function to UDF 
+convert_case_udf = udf(lambda p: convert_case(p))
+
+# use UDF in select stmt
+df.select(col("ID"), convert_case_udf(col("Name")).alias("Name")).show(truncate=False)
+
+# create a UDF function
+upper_case_udf = udf(lambda p: to_upper_case(p), StringType())    
+
+# Apply a UDF using withColumn
+df.withColumn("Upper Name", upper_case_udf(col("Name"))).show(truncate=False)
+
+# Using UDF on SQL 
+spark.udf.register("convert_UDF", convert_case, StringType())
+df.createOrReplaceTempView("NAME_TABLE")
+spark.sql("select ID, convert_UDF(Name) as Name from NAME_TABLE").show(truncate=False)
+     

code/bonus_chapters/UDF/scala/README.md

@@ -0,0 +1 @@
+Demo Spark's UDF (user-defined-function)

code/bonus_chapters/mappartitions/README.md

@@ -121,7 +121,9 @@ Now create a source `RDD[Integer]` and then apply  `mapPartitions()`:
 ~~~python
 >>> # spark : SparkSession object
 >>> data = [10, 20, 3, 4, 5, 2, 2, 20, 20, 10]
+>>> # rdd : RDD[integer]
 >>> rdd = spark.sparkContext.parallelize(data, 3)
+>>> # mapped : RDD[(integer, integer, integer)] : RDD[(count, min, max)]
 >>> mapped = rdd.mapPartitions(min_max)
 >>> mapped.collect()
 [(3, 3, 20), (3, 2, 5), (4, 2, 20)]
@@ -144,7 +146,9 @@ Note that you may perform final reduction by `RDD.reduce()` as well:
 ~~~python
 >>> # spark : SparkSession object
 >>> data = [10, 20, 3, 4, 5, 2, 2, 20, 20, 10]
+>>> # rdd : RDD[integer]
 >>> rdd = spark.sparkContext.parallelize(data, 3)
+>>> # mapped : RDD[(integer, integer, integer)] : RDD[(count, min, max)]
 >>> mapped = rdd.mapPartitions(min_max)
 >>> mapped.collect()
 [(3, 3, 20), (3, 2, 5), (4, 2, 20)]