#!/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 import.py Arguments"
        exit()
args_file = sys.argv[1]
hivedb = sys.argv[2]
domain = sys.argv[3]
port=sys.argv[4]
mysqldb=sys.argv[5]
username=sys.argv[6]
password=sys.argv[7]

def mysql_spark(table, hivedb, domain, port, mysqldb, username, password):

    print "*********************************************************table = {} ***************************".format(table)

    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")

    sqlContext.sql("create table {}.{} stored as parquet as select * from mytempTable".format(hivedb,table))

# file that contains table names
input = sc.textFile('/user/XXXXXXXX/mysql_spark/%s' %args_file).collect()

for table in input:
 mysql_spark(table, hivedb, domain, port, mysqldb, username, password)

sc.stop()