Initial Commit

Author: LearningJournal

.gitignore

@@ -2,3 +2,7 @@
 */.ipynb*
 *Untitled*
 
+*/.idea*
+*/app-logs*
+*__pycache__*
+

01-HelloSpark/HelloSpark.py

@@ -0,0 +1,29 @@
+import sys
+from pyspark.sql import *
+from lib.logger import Log4j
+from lib.utils import *
+
+if __name__ == "__main__":
+    conf = get_spark_app_config()
+
+    spark = SparkSession \
+        .builder \
+        .appName("HelloSpark") \
+        .master("local[2]") \
+        .getOrCreate()
+
+    logger = Log4j(spark)
+
+    if len(sys.argv) != 2:
+        logger.error("Usage: HelloSpark <filename>")
+        sys.exit(-1)
+
+    logger.info("Starting HelloSpark")
+
+    survey_raw_df = load_survey_df(spark, sys.argv[1])
+    partitioned_survey_df = survey_raw_df.repartition(2)
+    count_df = count_by_country(partitioned_survey_df)
+    count_df.show()
+
+    logger.info("Finished HelloSpark")
+    spark.stop()

01-HelloSpark/data/sample.csv

@@ -0,0 +1,10 @@
+"Timestamp","Age","Gender","Country","state","self_employed","family_history","treatment","work_interfere","no_employees","remote_work","tech_company","benefits","care_options","wellness_program","seek_help","anonymity","leave","mental_health_consequence","phys_health_consequence","coworkers","supervisor","mental_health_interview","phys_health_interview","mental_vs_physical","obs_consequence","comments"
+2014-08-27 11:29:31,37,"Female","United States","IL",NA,"No","Yes","Often","6-25","No","Yes","Yes","Not sure","No","Yes","Yes","Somewhat easy","No","No","Some of them","Yes","No","Maybe","Yes","No",NA
+2014-08-27 11:29:37,44,"M","United States","IN",NA,"No","No","Rarely","More than 1000","No","No","Don't know","No","Don't know","Don't know","Don't know","Don't know","Maybe","No","No","No","No","No","Don't know","No",NA
+2014-08-27 11:29:44,32,"Male","Canada",NA,NA,"No","No","Rarely","6-25","No","Yes","No","No","No","No","Don't know","Somewhat difficult","No","No","Yes","Yes","Yes","Yes","No","No",NA
+2014-08-27 11:29:46,31,"Male","United Kingdom",NA,NA,"Yes","Yes","Often","26-100","No","Yes","No","Yes","No","No","No","Somewhat difficult","Yes","Yes","Some of them","No","Maybe","Maybe","No","Yes",NA
+2014-08-27 11:30:22,31,"Male","United States","TX",NA,"No","No","Never","100-500","Yes","Yes","Yes","No","Don't know","Don't know","Don't know","Don't know","No","No","Some of them","Yes","Yes","Yes","Don't know","No",NA
+2014-08-27 11:31:22,33,"Male","United States","TN",NA,"Yes","No","Sometimes","6-25","No","Yes","Yes","Not sure","No","Don't know","Don't know","Don't know","No","No","Yes","Yes","No","Maybe","Don't know","No",NA
+2014-08-27 11:31:50,35,"Female","United States","MI",NA,"Yes","Yes","Sometimes","1-5","Yes","Yes","No","No","No","No","No","Somewhat difficult","Maybe","Maybe","Some of them","No","No","No","Don't know","No",NA
+2014-08-27 11:32:05,39,"M","Canada",NA,NA,"No","No","Never","1-5","Yes","Yes","No","Yes","No","No","Yes","Don't know","No","No","No","No","No","No","No","No",NA
+2014-08-27 11:32:39,42,"Female","United States","IL",NA,"Yes","Yes","Sometimes","100-500","No","Yes","Yes","Yes","No","No","No","Very difficult","Maybe","No","Yes","Yes","No","Maybe","No","No",NA

01-HelloSpark/lib/__init__.py

@@ -0,0 +1,19 @@
+class Log4j(object):
+    def __init__(self, spark):
+        root_class = "guru.learningjournal.spark.examples"
+        conf = spark.sparkContext.getConf()
+        app_name = conf.get("spark.app.name")
+        log4j = spark._jvm.org.apache.log4j
+        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)

01-HelloSpark/lib/logger.py

@@ -0,0 +1,27 @@
+import configparser
+
+from pyspark import SparkConf
+
+
+def load_survey_df(spark, data_file):
+    return spark.read \
+        .option("header", "true") \
+        .option("inferSchema", "true") \
+        .csv(data_file)
+
+
+def count_by_country(survey_df):
+    return survey_df.filter("Age < 40") \
+        .select("Age", "Gender", "Country", "state") \
+        .groupBy("Country") \
+        .count()
+
+
+def get_spark_app_config():
+    spark_conf = SparkConf()
+    config = configparser.ConfigParser()
+    config.read("spark.conf")
+
+    for (key, val) in config.items("SPARK_APP_CONFIGS"):
+        spark_conf.set(key, val)
+    return spark_conf

01-HelloSpark/lib/utils.py

@@ -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
+#log4j.appender.file.File=app-logs/hello-spark.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
+

01-HelloSpark/log4j.properties

@@ -0,0 +1 @@
+pyspark==2.4.5

01-HelloSpark/requirements.txt

@@ -0,0 +1,4 @@
+[SPARK_APP_CONFIGS]
+spark.app.name = HelloSpark
+spark.master = local[3]
+spark.sql.shuffle.partitions = 2

01-HelloSpark/spark.conf

@@ -0,0 +1,33 @@
+from unittest import TestCase
+from pyspark.sql import SparkSession
+from lib.utils import load_survey_df, count_by_country
+
+
+class UtilsTestCase(TestCase):
+    spark = None
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.spark = SparkSession.builder \
+            .master("local[3]") \
+            .appName("HelloSparkTest") \
+            .getOrCreate()
+
+    def test_datafile_loading(self):
+        sample_df = load_survey_df(self.spark, "data/sample.csv")
+        result_count = sample_df.count()
+        self.assertEqual(result_count, 9, "Record count should be 9")
+
+    def test_country_count(self):
+        sample_df = load_survey_df(self.spark, "data/sample.csv")
+        count_list = count_by_country(sample_df).collect()
+        count_dict = dict()
+        for row in count_list:
+            count_dict[row["Country"]] = row["count"]
+        self.assertEqual(count_dict["United States"], 4, "Count for United States should be 4")
+        self.assertEqual(count_dict["Canada"], 2, "Count for Canada should be 2")
+        self.assertEqual(count_dict["United Kingdom"], 1, "Count for Unites Kingdom should be 1")
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.spark.stop()