Initial Commit

Author: LearningJournal

07-RowDemo/RowDemo.py

@@ -0,0 +1,33 @@
+from pyspark.sql import *
+from pyspark.sql.functions import *
+from pyspark.sql.types import *
+
+from lib.logger import Log4j
+
+
+def to_date_df(df, fmt, fld):
+    return df.withColumn(fld, to_date(fld, fmt))
+
+
+if __name__ == "__main__":
+    spark = SparkSession \
+        .builder \
+        .master("local[3]") \
+        .appName("RowDemo") \
+        .getOrCreate()
+
+    logger = Log4j(spark)
+
+    my_schema = StructType([
+        StructField("ID", StringType()),
+        StructField("EventDate", StringType())])
+
+    my_rows = [Row("123", "04/05/2020"), Row("124", "4/5/2020"), Row("125", "04/5/2020"), Row("126", "4/05/2020")]
+    my_rdd = spark.sparkContext.parallelize(my_rows, 2)
+    my_df = spark.createDataFrame(my_rdd, my_schema)
+
+    my_df.printSchema()
+    my_df.show()
+    new_df = to_date_df(my_df, "M/d/y", "EventDate")
+    new_df.printSchema()
+    new_df.show()

07-RowDemo/RowDemo_Test.py

@@ -0,0 +1,35 @@
+from datetime import date
+from unittest import TestCase
+
+from pyspark.sql import *
+from pyspark.sql.types import *
+
+from RowDemo import to_date_df
+
+
+class RowDemoTestCase(TestCase):
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.spark = SparkSession.builder \
+            .master("local[3]") \
+            .appName("RowDemoTest") \
+            .getOrCreate()
+
+        my_schema = StructType([
+            StructField("ID", StringType()),
+            StructField("EventDate", StringType())])
+
+        my_rows = [Row("123", "04/05/2020"), Row("124", "4/5/2020"), Row("125", "04/5/2020"), Row("126", "4/05/2020")]
+        my_rdd = cls.spark.sparkContext.parallelize(my_rows, 2)
+        cls.my_df = cls.spark.createDataFrame(my_rdd, my_schema)
+
+    def test_data_type(self):
+        rows = to_date_df(self.my_df, "M/d/y", "EventDate").collect()
+        for row in rows:
+            self.assertIsInstance(row["EventDate"], date)
+
+    def test_date_value(self):
+        rows = to_date_df(self.my_df, "M/d/y", "EventDate").collect()
+        for row in rows:
+            self.assertEqual(row["EventDate"], date(2020, 4, 5))

07-RowDemo/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)

07-RowDemo/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
+