Initial Commit

Author: LearningJournal

18-OuterJoinDemo/OuterJoinDemo.py

@@ -0,0 +1,50 @@
+from pyspark.sql import SparkSession
+from pyspark.sql.functions import expr
+from lib.logger import Log4j
+
+if __name__ == "__main__":
+    spark = SparkSession \
+        .builder \
+        .appName("Spark Join Demo") \
+        .master("local[3]") \
+        .getOrCreate()
+
+    logger = Log4j(spark)
+
+    orders_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)]
+
+    order_df = spark.createDataFrame(orders_list).toDF("order_id", "prod_id", "unit_price", "qty")
+
+    product_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)]
+
+    product_df = spark.createDataFrame(product_list).toDF("prod_id", "prod_name", "list_price", "qty")
+
+    product_df.show()
+    order_df.show()
+
+    join_expr = order_df.prod_id == product_df.prod_id
+
+    product_renamed_df = product_df.withColumnRenamed("qty", "reorder_qty")
+
+    order_df.join(product_renamed_df, join_expr, "left") \
+        .drop(product_renamed_df.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()

18-OuterJoinDemo/lib/__init__.py

@@ -0,0 +1,21 @@
+class Log4j:
+    def __init__(self, spark):
+        log4j = spark._jvm.org.apache.log4j
+
+        root_class = "guru.learningjournal.spark.examples"
+        conf = spark.sparkContext.getConf()
+        app_name = conf.get("spark.app.name")
+
+        self.logger = log4j.LogManager.getLogger(root_class + "." + app_name)
+
+    def warn(self, message):
+        self.logger.warn(message)
+
+    def info(self, message):
+        self.logger.info(message)
+
+    def error(self, message):
+        self.logger.error(message)
+
+    def debug(self, message):
+        self.logger.debug(message)

18-OuterJoinDemo/lib/logger.py

@@ -0,0 +1,38 @@
+# 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.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
+