How to use the same spark context in a loop in Pyspark

Question

I have a shell script that invokes a pyspark job and creates tables in Hive. The script works fine. Now want to reduce the time taken for the script to run while invoking as cron jobs for more than 100 tables.

The import.sh script is below.

#!/bin/bash
source /home/$USER/source.sh

[ $# -ne 1 ] && { echo "Usage : $0  table "; exit 1; }

table=$1

TIMESTAMP=`date "+%Y-%m-%d"`
touch /home/$USER/logs/${TIMESTAMP}.success_log
touch /home/$USER/logs/${TIMESTAMP}.fail_log
success_logs=/home/$USER/logs/${TIMESTAMP}.success_log
failed_logs=/home/$USER/logs/${TIMESTAMP}.fail_log

function log_status
{
       status=$1
       message=$2
       if [ "$status" -ne 0 ]; then
                echo "`date +\"%Y-%m-%d %H:%M:%S\"` [ERROR] $message [Status] $status : failed" | tee -a "${failed_logs}"
                #echo "Please find the attached log file for more details"
                exit 1
                else
                    echo "`date +\"%Y-%m-%d %H:%M:%S\"` [INFO] $message [Status] $status : success" | tee -a "${success_logs}"
                fi
}

#Executing Spark job with python script
spark-submit --name "SparkJob" --driver-memory 2g --executor-memory 4g --num-executors 3 --conf "spark.yarn.executor.memoryOverhead=709" --conf "spark.yarn.driver.memoryOverhead=209"   /home/$USER/import.py ${table} ${mysqldb} ${hivedb} ${domain} ${port} ${username} ${password}

g_STATUS=$?
log_status $g_STATUS "SPARKJOB ${table} EXECUTION"

echo "*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-*-"

The import.py pyspark script is below

#!/usr/bin/env python
import sys
from pyspark import SparkContext, SparkConf
from pyspark.sql import HiveContext
conf = SparkConf()
sc = SparkContext(conf=conf)
sqlContext = HiveContext(sc)

#Condition to specify exact number of arguments in the spark-submit command line
if len(sys.argv) != 8:
     print "Invalid number of args......"
     print "Usage: spark-submit file.py Input_path Output_path"
     exit()
table=sys.argv[1]
mysqldb=sys.argv[2]
hivedb=sys.argv[3]
domain=sys.argv[4]
port=sys.argv[5]
username=sys.argv[6]
password=sys.argv[7]

#Get table from MYSQL as a Data Frame into Spark
df = sqlContext.read.format("jdbc").option("url", "{}:{}/{}".format(domain,port,mysqldb)).option("driver", "com.mysql.jdbc.Driver").option("dbtable","{}".format(table)).option("user", "{}".format(username)).option("password", "{}".format(password)).load()

df.registerTempTable('mytempTable')

# Create Table in Hive using the temptable
sqlContext.sql("create table {}.`{}` as select * from mytempTable".format(hivedb,table))

sc.stop()

I am getting the desired results with my above scripts.

I am able to get data from MYSQL database to PYSPARK and then able to create tables in HIVE. Also I am able to collect the logs for each sparkjob.

For example I have 100 tables to import from MYSQL to HIVE.

I am scheduling this shell script as a cronjob. This shell script will run 10 times in parallel in a loop.

I mean ./import.sh table1, ./import.sh table2,./import.sh table3,./import.sh table4 and so on 10 times

Now my question

Each time the shell script is invoked a spark context is being opened. It takes around 10-12 seconds for a spark context to open. Now at the end we are closing the spark context using sc.stop().

So If I want to run the script I have to open 100 spark context and close them after completing the job.

What I want is instead of closing the spark context after completion of the job, Can we reuse the same sparkContext for another table in loop.

I mean when I do this ./import.sh table1 I want to use the sparkcontext that was opened in this job for the table that comes next in the loop. I mean for the table 11.

Is this possible?

If yes then How can I achieve that?

How to use the same spark context in a loop in Pyspark

1 Answers