Initial Commit

Author: LearningJournal

SBDL - Starter/.gitignore

@@ -0,0 +1,5 @@
+# Default ignored files
+/workspace.xml
+.idea/*
+lib/__pycache__/*
+

SBDL - Starter/Jenkinsfile

@@ -0,0 +1,40 @@
+pipeline {
+    agent any
+
+    stages {
+        stage('Build') {
+            steps {
+               sh 'pipenv --python python3 sync'
+            }
+        }
+        stage('Test') {
+            steps {
+               sh 'pipenv run pytest'
+            }
+        }
+        stage('Package') {
+	    when{
+		    anyOf{ branch "master" ; branch 'release' }
+	    }
+            steps {
+               sh 'zip -r sbdl.zip lib'
+            }
+        }
+	stage('Release') {
+	   when{
+	      branch 'release'
+	   }
+           steps {
+              sh "scp -i /home/prashant/cred/edge-node_key.pem -o 'StrictHostKeyChecking no' -r sbdl.zip log4j.properties sbdl_main.py sbdl_submit.sh conf [email protected]:/home/prashant/sbdl-qa"
+           }
+        }
+	stage('Deploy') {
+	   when{
+	      branch 'master'
+	   }
+           steps {
+               sh "scp -i /home/prashant/cred/edge-node_key.pem -o 'StrictHostKeyChecking no' -r sbdl.zip log4j.properties sbdl_main.py sbdl_submit.sh conf [email protected]:/home/prashant/sbdl-prod"
+           }
+        }
+    }
+}

SBDL - Starter/Pipfile

@@ -0,0 +1,14 @@
+[[source]]
+url = "https://pypi.org/simple"
+verify_ssl = true
+name = "pypi"
+
+[packages]
+pyspark = "*"
+pytest = "*"
+
+[dev-packages]
+
+[requires]
+python_version = "3.10"
+

SBDL - Starter/conf/sbdl.conf

@@ -0,0 +1,13 @@
+[LOCAL]
+enable.hive = false
+hive.database = null
+kafka.topic = sbdl_kafka_cloud
+[QA]
+enable.hive = true
+hive.database = sbdl_db_qa
+kafka.topic = sbdl_kafka_qa
+[PROD]
+enable.hive = true
+hive.database = sbdl_db
+kafka.topic = sbdl_kafka
+

SBDL - Starter/conf/spark.conf

@@ -0,0 +1,20 @@
+[LOCAL]
+spark.app.name = sbdl-local
+spark.executor.instances = 2
+spark.executor.cores = 1
+spark.executor.memory = 1G
+spark.sql.shuffle.partitions = 5
+[QA]
+spark.app.name = sbdl-qa
+spark.executor.instances = 2
+spark.executor.cores = 1
+spark.executor.memory = 4G
+spark.sql.shuffle.partitions = 1000
+[PROD]
+spark.app.name = sbdl
+spark.executor.instances = 2
+spark.executor.cores = 1
+spark.executor.memory = 4G
+spark.sql.shuffle.partitions = 1000
+
+

SBDL - Starter/lib/Utils.py

@@ -0,0 +1,17 @@
+from pyspark.sql import SparkSession
+
+
+def get_spark_session(env):
+    if env == "LOCAL":
+        return SparkSession.builder \
+            .config('spark.driver.extraJavaOptions',
+                    '-Dlog4j.configuration=file:log4j.properties') \
+            .master("local[2]") \
+            .enableHiveSupport() \
+            .getOrCreate()
+    else:
+        return SparkSession.builder \
+            .enableHiveSupport() \
+            .getOrCreate()
+
+

SBDL - Starter/lib/__init__.py

@@ -0,0 +1,18 @@
+class Log4j(object):
+    def __init__(self, spark):
+        log4j = spark._jvm.org.apache.log4j
+        self.logger = log4j.LogManager.getLogger("sbdl")
+
+    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)
+
+

SBDL - Starter/lib/logger.py

@@ -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.sbdl=INFO, console, file
+log4j.additivity.sbdl=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
+

SBDL - Starter/log4j.properties

@@ -0,0 +1,18 @@
+import sys
+
+from lib import Utils
+from lib.logger import Log4j
+
+if __name__ == '__main__':
+
+    if len(sys.argv) < 3:
+        print("Usage: sbdl {local, qa, prod} {load_date} : Arguments are missing")
+        sys.exit(-1)
+
+    job_run_env = sys.argv[1].upper()
+    load_date = sys.argv[2]
+
+    spark = Utils.get_spark_session(job_run_env)
+    logger = Log4j(spark)
+
+    logger.info("Finished creating Spark Session")