Initial Commit

Author: LearningJournal

Diff for: 19-OuterJoinDemo/build.sbt

@@ -0,0 +1,15 @@
+name := "OuterJoinDemo"
+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,
+  "org.apache.spark" %% "spark-hive" % sparkVersion
+)
+
+libraryDependencies ++= sparkDependencies

Diff for: 19-OuterJoinDemo/log4j.properties

@@ -0,0 +1,40 @@
+# 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, file
+log4j.additivity.guru.learningjournal.spark.examples=false
+
+#define rolling file appender
+log4j.appender.file=org.apache.log4j.RollingFileAppender
+log4j.appender.file.File=${spark.yarn.app.container.log.dir}/${logfile.name}.log
+#define following in Java System
+# -Dlog4j.configuration=file:log4j.properties
+# -Dlogfile.name=hello-spark
+# -Dspark.yarn.app.container.log.dir=app-logs
+log4j.appender.file.ImmediateFlush=true
+log4j.appender.file.Append=false
+log4j.appender.file.MaxFileSize=500MB
+log4j.appender.file.MaxBackupIndex=2
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+log4j.appender.file.layout.conversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
+
+
+# Recommendations from Spark template
+log4j.logger.org.apache.hadoop=ERROR
+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
+

Diff for: 19-OuterJoinDemo/src/main/scala/guru/learningjournal/spark/examples/OuterJoinDemo.scala

@@ -0,0 +1,56 @@
+package guru.learningjournal.spark.examples
+
+import org.apache.log4j.Logger
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.functions._
+
+object OuterJoinDemo extends Serializable{
+  @transient lazy val logger: Logger = Logger.getLogger(getClass.getName)
+
+  def main(args: Array[String]): Unit = {
+
+    val spark = SparkSession.builder()
+      .appName("Spark Join Demo")
+      .master("local[3]")
+      .getOrCreate()
+
+    val ordersList = List(
+      ("01", "02", 350, 1),
+      ("01", "04", 580, 1),
+      ("01", "07", 320, 2),
+      ("02", "03", 450, 1),
+      ("02", "06", 220, 1),
+      ("03", "01", 195, 1),
+      ("04", "09", 270, 3),
+      ("04", "08", 410, 2),
+      ("05", "02", 350, 1)
+    )
+    val orderDF = spark.createDataFrame(ordersList).toDF("order_id", "prod_id", "unit_price", "qty")
+
+    val productList = List(
+      ("01", "Scroll Mouse", 250, 20),
+      ("02", "Optical Mouse", 350, 20),
+      ("03", "Wireless Mouse", 450, 50),
+      ("04", "Wireless Keyboard", 580, 50),
+      ("05", "Standard Keyboard", 360, 10),
+      ("06", "16 GB Flash Storage", 240, 100),
+      ("07", "32 GB Flash Storage", 320, 50),
+      ("08", "64 GB Flash Storage", 430, 25)
+    )
+    val productDF = spark.createDataFrame(productList).toDF("prod_id", "prod_name", "list_price", "qty")
+
+    val joinExpr = orderDF.col("prod_id") === productDF.col("prod_id")
+    val joinType = "left"
+
+    val productRenamedDF = productDF.withColumnRenamed("qty", "reorder_qty")
+
+    orderDF.join(productRenamedDF, joinExpr, joinType)
+      .drop(productDF.col("prod_id"))
+      .select("order_id", "prod_id", "prod_name", "unit_price", "list_price", "qty")
+      .withColumn("prod_name", expr("coalesce(prod_name, prod_id)"))
+      .withColumn("list_price", expr("coalesce(list_price, unit_price)"))
+      .sort("order_id")
+      .show()
+  }
+
+}