| import traceback | |
| from pyspark.sql import SparkSession | |
| from pyspark import SparkConf | |
| from pyspark.sql.functions import col,regexp_replace, concat_ws, when, collect_list, lit, to_timestamp | |
| from pyspark.sql.functions import year, month, date_format | |
| from pyspark.sql import functions as F | |
| from pyspark.sql.types import LongType,DecimalType,IntegerType,TimestampType,DoubleType | |
| from pyspark.sql.functions import * | |
| from pytz import timezone | |
| from datetime import datetime,timedelta | |
| from pyspark.sql.window import Window | |
| import json | |
| import sys | |
| import logging | |
| import datetime | |
| import time | |
| import os | |
| import psycopg2 | |
| import requests | |
| from requests.auth import HTTPBasicAuth | |
| import base64 | |
| import functools | |
| import boto3 | |
| # adding '/home/hadoop' path of emr master instance as our downloaded packages will be present at this path | |
| sys.path.append('/home/hadoop') | |
| curr_time = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') | |
| log_file_name = 'job_' + str(datetime.datetime.now().strftime('%Y%m%d_%H%M%S_%f')) + '.log' | |
| extra = {'log_file_name': log_file_name} | |
| logger = logging.getLogger(__name__) | |
| syslog = logging.FileHandler(log_file_name, mode='w') | |
| formatter = logging.Formatter('%(log_file_name)s;%(asctime)s;%(levelname)s;%(message)s') | |
| syslog.setFormatter(formatter) | |
| logger.setLevel(logging.INFO) | |
| logger.addHandler(syslog) | |
| logger = logging.LoggerAdapter(logger, extra) | |
| def read_config(config_path): | |
| logger.info("Inside read config") | |
| try: | |
| # checking if config path provided as input is s3 path or file system path | |
| if config_path[0:2] == 's3': | |
| # read config file from s3 | |
| logger.info("Reading config file from S3") | |
| s3 = boto3.resource('s3') | |
| file_object = s3.Object(config_path.split('/')[2], '/'.join(config_path.split('/')[3:])) | |
| file_content = file_object.get()['Body'].read().decode('utf-8') | |
| # converting file content to json format | |
| json_content = json.loads(file_content) | |
| json_object = json.dumps(json_content) | |
| else: | |
| # reading config file from system | |
| logger.info("Reading config file from path : " + config_path) | |
| # converting file content to json format | |
| json_content = json.load(open(config_path, 'r')) | |
| json_object = json.dumps(json_content) | |
| logger.info("Input Config Details:") | |
| logger.info(json_object) | |
| return json_content | |
| except Exception as e: | |
| raise Exception("Error reading config.") | |
| def get_secret(secret): | |
| secret_name = secret | |
| region_name = "ap-south-1" | |
| session = boto3.session.Session() | |
| client = session.client( | |
| service_name='secretsmanager', | |
| region_name=region_name, | |
| ) | |
| try: | |
| get_secret_value_response = client.get_secret_value(SecretId=secret_name) | |
| except ClientError as e: | |
| if e.response['Error']['Code'] == 'ResourceNotFoundException': | |
| print("The requested secret " + secret_name + " was not found") | |
| elif e.response['Error']['Code'] == 'InvalidRequestException': | |
| print("The request was invalid due to:", e) | |
| elif e.response['Error']['Code'] == 'InvalidParameterException': | |
| print("The request had invalid params:", e) | |
| elif e.response['Error']['Code'] == 'DecryptionFailure': | |
| print("The requested secret can't be decrypted using the provided KMS key:", e) | |
| elif e.response['Error']['Code'] == 'InternalServiceError': | |
| print("An error occurred on service side:", e) | |
| else: | |
| # Secrets Manager decrypts the secret value using the associated KMS CMK | |
| # Depending on whether the secret was a string or binary, only one of these fields will be populated | |
| if 'SecretString' in get_secret_value_response: | |
| text_secret_data = get_secret_value_response['SecretString'] | |
| return text_secret_data | |
| else: | |
| binary_secret_data = get_secret_value_response['SecretBinary'] | |
| return binary_secret_data | |
| logger.info("Secret manager read complete") | |
| def create_spark_session(config): | |
| logger.info("Inside create spark session") | |
| try: | |
| conf = SparkConf() | |
| # setting spark configuration properties provided in config file | |
| spark_conf = dict(config['spark_properties']) | |
| for key in spark_conf.keys(): | |
| conf.set(key, spark_conf[key]) | |
| logger.info("Secret manager read") | |
| if 'application_name' in list(config.keys()): | |
| if config['application_name'] != '': | |
| app_name = config['application_name'] | |
| else: | |
| app_name = 'DefaultApp' | |
| else: | |
| app_name = 'DefaultApp' | |
| logger.info("Secret manager read start") | |
| # creating spark session | |
| spark = SparkSession.builder.config(conf=conf).appName(app_name).enableHiveSupport().getOrCreate() | |
| spark.sparkContext.setLogLevel("ERROR") | |
| spark.conf.set("spark.sql.autoBroadcastJoinThreshold",-1) | |
| spark.conf.set("spark.sql.legacy.parquet.datetimeRebaseModeInRead",'LEGACY') | |
| spark.conf.set("spark.sql.legacy.timeParserPolicy",'CORRECTED') | |
| spark.conf.set("spark.sql.legacy.parquet.int96RebaseModeInWrite",'CORRECTED') | |
| spark.conf.set("spark.sql.legacy.parquet.datetimeRebaseModeInWrite",'CORRECTED') | |
| spark.conf.set("spark.sql.legacy.parquet.int96RebaseModeInRead",'CORRECTED') | |
| spark.conf.set("spark.sql.shuffle.partitions",100) | |
| logger.info("Spark session object created") | |
| return spark | |
| except Exception as e: | |
| raise Exception("Error in Spark Session Creation.") | |
| def read_file(spark,config,table): | |
| readOptions = { | |
| 'hoodie.datasource.query.type': 'incremental', | |
| 'hoodie.datasource.hive_sync.support_timestamp': 'true' | |
| } | |
| path = config['Paths'][table] | |
| df=spark.read.format("hudi").load(path) | |
| df =df.withColumn('_hoodie_commit_time',to_timestamp(F.concat(F.substring(col('_hoodie_commit_time'),1,4),F.lit('-'),\ | |
| F.substring(col('_hoodie_commit_time'),5,2),F.lit('-'),\ | |
| F.substring(col('_hoodie_commit_time'),7,2),F.lit(' '),\ | |
| F.substring(col('_hoodie_commit_time'),9,2),F.lit(':'),\ | |
| F.substring(col('_hoodie_commit_time'),11,2),F.lit(':'),\ | |
| F.substring(col('_hoodie_commit_time'),13,2)\ | |
| ))) | |
| return df | |
| def get_max_audit_batch(conn,job_name, config): | |
| cur = conn.cursor() | |
| cur.execute("SELECT COALESCE(MAX(COALESCE(BATCH_ID,0)),0)+1 FROM "+config['audit_table']) | |
| result = cur.fetchall()[0][0] | |
| logger.info("Maximum batch id in Audit Table is :"+str(result)) | |
| return result | |
| def read_max_update_date(conn, job_name, table, config): | |
| try: | |
| cur = conn.cursor() | |
| cur.execute("SELECT MAX(max_update_date) from "+config['audit_table']+" WHERE mart_table_name = '"+job_name+"' AND src_table_name = '"+table+"'") | |
| query_results = cur.fetchall() | |
| except Exception as e: | |
| print("Database connection failed due to {}".format(e)) | |
| raise Exception("Error reading audit table.") | |
| return query_results | |
| logger.info("Reading max of max_update_date from audit table complete") | |
| def insert_max_update_date(spark,conn, job_name, table, max_update_date,source_reference_date, max_batch_id, config): | |
| try: | |
| cur = conn.cursor() | |
| cur.execute("INSERT INTO "+config['audit_table']+"(mart_table_name, src_table_name, max_update_date, load_timestamp,source_reference_date,batch_id) VALUES ('"+str(job_name)+"', '"+str(table)+"', '"+str(max_update_date)+"', SYSDATE ,'"+str(source_reference_date)+"' as source_reference_date,cast('"+str(max_batch_id)+"' as int) as batch_id)") | |
| except Exception as e: | |
| print("Database connection failed due to {}".format(e)) | |
| raise Exception("Error Updating audit table.") | |
| logger.info("Inserting max max_update_date into audit table complete") | |
| def write_file(spark,conn,redshift_iam_role,resultdf_path, config, table_name): | |
| #Writing resultant data into incr table using copy command | |
| logger.info("write data to redshift started") | |
| try: | |
| cur = conn.cursor() | |
| cur.execute(f"""Truncate table int.{table_name};commit;""" ) | |
| sql="""COPY %s FROM '%s' credentials 'aws_iam_role=%s' FORMAT PARQUET; commit;""" % \ | |
| (f"int.{table_name}", resultdf_path,redshift_iam_role) | |
| cur.execute(sql) | |
| except Exception as e: | |
| print("Database connection failed due to {}".format(e)) | |
| raise Exception("Error Inserting target table.") | |
| print("write complete") | |
| logger.info("upsert data to rds completed") | |
| def main(): | |
| logger.info("Inside main function") | |
| if len(sys.argv) != 2: | |
| logger.info(len(sys.argv)) | |
| logger.info("Command line arguments : " + str(sys.argv)) | |
| logger.info("Incorrect command line arguments.") | |
| exit(1) | |
| config = {} | |
| spark = '' | |
| job_status = '' | |
| try: | |
| # reading json config file | |
| logger.info("Calling function to read config file") | |
| config = read_config(sys.argv[1]) | |
| logger.info("Calling function to create Spark session object") | |
| #creating spark session | |
| spark = create_spark_session(config) | |
| logger.info("Calling function to read input file") | |
| start_time = datetime.datetime.now(timezone("Asia/Kolkata")).strftime('%Y-%m-%d %H:%M:%S') | |
| #creating redshift database connection | |
| redshift_secret = get_secret(config['redshift_secret']) | |
| redshift_secret = json.loads(redshift_secret) | |
| redshift_user = redshift_secret['username'] | |
| redshift_pwd = redshift_secret['password'] | |
| redshift_host = redshift_secret['host'] | |
| redshift_port = str(redshift_secret['port']) | |
| redshift_dbname = redshift_secret['dbname'] | |
| #creating database connection | |
| redshift_conn=psycopg2.connect(dbname=redshift_dbname, host=redshift_host, port=redshift_port, user=redshift_user, password=redshift_pwd) | |
| redshift_dburl = "jdbc:postgresql://"+redshift_host+":"+redshift_port+"/"+redshift_dbname | |
| cur = redshift_conn.cursor() | |
| max_batch_id = get_max_audit_batch(redshift_conn, config['application_name'], config) | |
| INSERT_CODE_1 | |
| #writing from parquet to table in database | |
| write_file(spark, redshift_conn, config['redshift_iam_role'],config['incr2df_path'],config, config['incr2df']) | |
| write_file(spark, redshift_conn, config['redshift_iam_role'],config['resultdf_path'],config, config['resultdf']) | |
| INSERT_CODE_2 | |
| print('Run Successful') | |
| print('End of Code') | |
| except Exception as e: | |
| #job gets error | |
| job_status = 'Failed' | |
| print(e) | |
| finally: | |
| spark.catalog.clearCache() | |
| redshift_conn.commit() | |
| redshift_conn.close() | |
| spark.stop() | |
| if __name__ == "__main__": | |
| # calling main function | |
| logger.info("Calling main function") | |
| main() |