Initial Commit

Author: LearningJournal

.gitignore

@@ -10,4 +10,5 @@
 # Leave target directory
 */target/*
 */metastore_db*
-*/spark-warehouse*
+*/spark-warehouse*
+*/output*

21-ComplexTypesDemo/build.sbt

@@ -0,0 +1,13 @@
+name := "ComplexTypesDemo"
+organization := "guru.learningjournal"
+version := "0.1"
+scalaVersion := "2.12.10"
+autoScalaLibrary := false
+val sparkVersion = "3.0.0"
+
+val sparkDependencies = Seq(
+  "org.apache.spark" %% "spark-core" % sparkVersion,
+  "org.apache.spark" %% "spark-sql" % sparkVersion
+)
+
+libraryDependencies ++= sparkDependencies

21-ComplexTypesDemo/data/Invoice-set1.json

@@ -0,0 +1,27 @@
+# Set everything to be logged to the console
+log4j.rootCategory=WARN, console
+
+# define console appender
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.target=System.out
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
+
+#application log
+log4j.logger.guru.learningjournal.spark.examples=INFO, console
+log4j.additivity.guru.learningjournal.spark.examples=false
+
+#define following in Java System
+# -Dlog4j.configuration=file:log4j.properties
+
+# Recommendations from Spark template
+log4j.logger.org.apache.spark.repl.Main=WARN
+log4j.logger.org.spark_project.jetty=WARN
+log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycle=ERROR
+log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=INFO
+log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=INFO
+log4j.logger.org.apache.parquet=ERROR
+log4j.logger.parquet=ERROR
+log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
+log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR
+

21-ComplexTypesDemo/data/Invoice-set2.json

@@ -0,0 +1,66 @@
+package guru.learningjournal.spark.examples
+
+import org.apache.log4j.Logger
+import org.apache.spark.sql.{SaveMode, SparkSession}
+import org.apache.spark.sql.functions.expr
+import org.apache.spark.sql.types.{ArrayType, DoubleType, IntegerType, LongType, StringType, StructField, StructType}
+
+object ComplexTypesDemo extends Serializable {
+  @transient lazy val logger: Logger = Logger.getLogger(getClass.getName)
+
+  def main(args: Array[String]): Unit = {
+    val spark = SparkSession.builder()
+      .master("local[3]")
+      .appName("Complex Types Demo")
+      .getOrCreate()
+
+    val schema = StructType(List(
+      StructField("InvoiceNumber", StringType),
+      StructField("CreatedTime", LongType),
+      StructField("StoreID", StringType),
+      StructField("PosID", StringType),
+      StructField("CashierID", StringType),
+      StructField("CustomerType", StringType),
+      StructField("CustomerCardNo", StringType),
+      StructField("TotalAmount", DoubleType),
+      StructField("NumberOfItems", IntegerType),
+      StructField("PaymentMethod", StringType),
+      StructField("CGST", DoubleType),
+      StructField("SGST", DoubleType),
+      StructField("CESS", DoubleType),
+      StructField("DeliveryType", StringType),
+      StructField("DeliveryAddress", StructType(List(
+        StructField("AddressLine", StringType),
+        StructField("City", StringType),
+        StructField("State", StringType),
+        StructField("PinCode", StringType),
+        StructField("ContactNumber", StringType)
+      ))),
+      StructField("InvoiceLineItems", ArrayType(StructType(List(
+        StructField("ItemCode", StringType),
+        StructField("ItemDescription", StringType),
+        StructField("ItemPrice", DoubleType),
+        StructField("ItemQty", IntegerType),
+        StructField("TotalValue", DoubleType),
+      )))),
+    ))
+
+    val df1 = spark.read.schema(schema).json("data/")
+
+    val df2 = df1.selectExpr("InvoiceNumber", "CreatedTime", "StoreID", "PosID",
+      "CustomerType", "PaymentMethod", "DeliveryType",
+      "DeliveryAddress.City", "DeliveryAddress.State", "DeliveryAddress.PinCode",
+      "explode(InvoiceLineItems) as LineItem")
+
+    val df4 = df2.withColumn("ItemCode", expr("LineItem.ItemCode"))
+      .withColumn("ItemDescription", expr("LineItem.ItemDescription"))
+      .withColumn("ItemPrice", expr("LineItem.ItemPrice"))
+      .withColumn("ItemQty", expr("LineItem.ItemQty"))
+      .withColumn("TotalValue", expr("LineItem.TotalValue"))
+      .drop("LineItem")
+
+    df4.write.mode(SaveMode.Overwrite).parquet("output/")
+    logger.info("Finish writing data.")
+  }
+
+}