import com.amazonaws.services.glue.{DynamicFrame, GlueContext}
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import java.util.Calendar
import org.apache.spark.SparkContext
import org.apache.spark.sql.functions._
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Row
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.{IntegerType, StringType, TimestampType }
import scala.collection.JavaConverters._


object CreateCustomerAccount {

  var jdbcUrl : String = _
  var dbUser : String = _
  var dbPassword : String = _
  var catalogueConnection : String = _
  var catalogueDatabase : String = _
  var glueSourceDatabase : String = _
  var glueSourceTable : String = _
  var tempStorage : String = _
  var connectionType : String = _

  def getConnectionOptionsForTable(table : String) : Map[String, String] = {
      Map(
          "url"-> jdbcUrl, 
          "user"-> dbUser,
          "database" -> catalogueDatabase,
          "dbtable" -> table,
          "password"-> dbPassword,
          "dbtable"-> table, 
          "redshiftTmpDir"-> tempStorage
      )
  }

  def getNewEntries(existingDataFrame : DataFrame, inputDataFrame : DataFrame, columns : List[String]) : DataFrame = {
    var firstColumn = columns(0)
    var newEntries = existingDataFrame.alias("e") 
      .join(inputDataFrame.alias("i"), 
            columns, 
            "right") 
      .filter(col("e." + firstColumn).isNull) 
      .select("i.*") 
    return newEntries
  }

  def main(sysArgs: Array[String]) {


    println("STARTING")
    val sc: SparkContext = new SparkContext()
    val glueContext: GlueContext = new GlueContext(sc)
    val sparkSession: SparkSession = glueContext.getSparkSession
    import sparkSession.implicits._

    val args = GlueArgParser.getResolvedOptions(sysArgs, 
        Seq(
            "JOB_NAME",
            "JDBC_URL",
            "DB_USER",
            "DB_PASSWORD",
            "CATALOGUE_CONNECTION",
            "CATALOGUE_DATABASE",
            "GLUE_SOURCE_DATABASE",
            "GLUE_SOURCE_TABLE",
            "TEMP_STORAGE",
            "CONNECTION_TYPE"
        ).toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)

    this.jdbcUrl = args("JDBC_URL")
    this.dbUser = args("DB_USER")
    this.dbPassword = args("DB_PASSWORD")
    this.catalogueConnection = args("CATALOGUE_CONNECTION")
    this.catalogueDatabase = args("CATALOGUE_DATABASE")
    this.glueSourceDatabase = args("GLUE_SOURCE_DATABASE")
    this.glueSourceTable = args("GLUE_SOURCE_TABLE")
    this.tempStorage = args("TEMP_STORAGE")
    this.connectionType = args("CONNECTION_TYPE")
    
    val businessAccountsTable = "business_account"
    val parentBusinessAccountsTable = "parent_business_account"
    val locationsTable = "location"
    val addressesTable = "address"
    val phoneNumbersTable = "phone_number"
    val emailsTable = "email"
    val placeOfBirthTable = "place_of_birth"
    val customerAccountsTable = "customer_account"
    val customerAccountsLocationsTable = "customer_account_location"
    val customerAccountsPhoneNumbersTable = "customer_account_phone_number"
    val customerAccountsEmailsTable = "customer_account_email"
    val customerAccountsPlaceOfBirthTable = "customer_account_place_of_birth"
    val parentBAChildCustomerAccountTable = "parent_business_account_child_customer_account"
    val bAChildCustomerAccountTable = "business_account_child_customer_account"
    val rootTableName = "general_customer_account"
    
    val businessAccountsKeys : List[String]= List("id")
    val parentBusinessAccountsKeys : List[String] = List("id")
    val addressesKeys : List[String] = List("id")
    val locationsKeys : List[String] = List("id")
    val phoneNumbersKeys : List[String] = List("id")
    val emailsKeys : List[String] = List("id")
    val placeOfBirthKeys : List[String] = List("id")
    val customerAccountsKeys : List[String] = List("id")
    val customerAccounts_locationsKeys : List[String] = List("customer_account_id", "location_id")
    val customerAccounts_phoneNumbersKeys : List[String] = List("customer_account_id", "phone_number_id")
    val customerAccounts_emailsKeys : List[String] = List("customer_account_id", "email_id")
    val customerAccounts_placeOfBirthKeys : List[String] = List("customer_account_id", "place_of_birth_id")
    val parentBA_childCustomerAccountKeys : List[String] = List("parent_business_account_id", "child_customer_account_id")
    val bA_childCustomerAccountKeys : List[String] = List("business_account_id", "child_customer_account_id")

    val businessAccountConnOptions = getConnectionOptionsForTable(businessAccountsTable)
    val parentBusinessAccountConnOptions = getConnectionOptionsForTable(parentBusinessAccountsTable)
    var addressConnOptions = getConnectionOptionsForTable(addressesTable)
    var locationConnOptions = getConnectionOptionsForTable(locationsTable)
    var phoneNumberConnOptions = getConnectionOptionsForTable(phoneNumbersTable)
    var emailConnOptions = getConnectionOptionsForTable(emailsTable)
    var placeOfBirthConnOptions = getConnectionOptionsForTable(placeOfBirthTable)
    var customerAccountConnOptions = getConnectionOptionsForTable(customerAccountsTable)
    var customerAccountLocationConnOptions = getConnectionOptionsForTable(customerAccountsLocationsTable)
    var customerAccountEmailConnOptions = getConnectionOptionsForTable(customerAccountsEmailsTable)
    var customerAccountPhoneNumberConnOptions = getConnectionOptionsForTable(customerAccountsPhoneNumbersTable)
    var customerAccountPlaceOfBirthConnOptions = getConnectionOptionsForTable(customerAccountsPlaceOfBirthTable)
    var businessAccountChildCustomerConnOptions = getConnectionOptionsForTable(businessAccountsChildCustomersTable)
    var parentBusinessAccountChildCustomerConnOptions = getConnectionOptionsForTable(parentBusinessAccountsChildCustomersTable)
    println("ETL Job: Done preparing connection_options")
    
    //get existing records as spark data frames
    var existingBusinessAccounts = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(businessAccountConnOptions)).getDataFrame
    var existingParentBusinessAccounts = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(parentBusinessAccountConnOptions)).getDataFrame
    var existingAddresses = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(addressConnOptions)).getDataFrame
    var existingLocations = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(locationConnOptions)).getDataFrame
    var existingPhoneNumbers = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(phoneNumberConnOptions)).getDataFrame
    var existingEmails = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(emailConnOptions)).getDataFrame
    var existingPlaceOfBirths = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(placeOfBirthConnOptions)).getDataFrame
    var existingCustomerAccounts = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(customerAccountConnOptions)).getDataFrame
    var existingCustomerAccountsLocations = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(customerAccountLocationConnOptions)).getDataFrame
    var existingCustomerAccountsEmails = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(customerAccountEmailConnOptions)).getDataFrame
    var existingCustomerAccountsPlaceOfBirths = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(customerAccountPlaceOfBirthConnOptions)).getDataFrame
    var existingCustomerAccountsPhoneNumbers = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(customerAccountPhoneNumberConnOptions)).getDataFrame
    var existingBusinessAccountChildCustomers = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(businessAccountChildCustomerConnOptions)).getDataFrame
    var existingParentBusinessAccountChildCustomers = glueContext.getSource(connectionType = connectionType, connectionOptions = JsonOptions(parentBusinessAccountChildCustomerConnOptions)).getDataFrame
    print("ETL Job: Done getting existing records")
    
    //get input data as DynamicFrame
    val customerDyf = glueContext.getCatalogSource(database = glueSourceDatabase, tableName = glueSourceTable).getDynamicFrame()
    
    //relationalize dynamic frame into separate DATA frames (spark)
    val relationalizedCustomer = customerDyf.relationalize(rootTableName, tempStorage)
    
    var customerAccountsDF = relationalizedCustomer.find(_.name.equals("general_customer_account")).get.toDF()
    var addressesDF = relationalizedCustomer.find(_.name.equals("general_customer_account_locations.val.addresses")).get.toDF()
    var emailsDF = relationalizedCustomer.find(_.name.equals("general_customer_account_emails")).get.toDF()
    var locationsDF = relationalizedCustomer.find(_.name.equals("general_customer_account_locations")).get.toDF()
    var phoneNumbersDF = relationalizedCustomer.find(_.name.equals("general_customer_account_phoneNumbers")).get.toDF()
    println("ETL Job: Done Relationalizing")

    //create Place Of Birth table
    var placeOfBirthDF = customerAccountsDF.select(
      col("id").as("tempCustomerAccountId"),
      col("placeOfBirth.id").as("id"),
      col("placeOfBirth.countryCode").as("country_code"),
      col("placeOfBirth.stateCode").as("state_code"),
      col("placeOfBirth.cityName").as("city"))
      

    //need to rename some columns to snake_case
    customerAccountsDF = customerAccountsDF
      .withColumnRenamed("userStatus", "user_status") 
      .withColumnRenamed("businessAccountId", "business_account_id") 
      .withColumnRenamed("customerNumber", "customer_number") 
      .withColumnRenamed("firstName", "first_name") 
      .withColumnRenamed("middleName", "middle_name") 
      .withColumnRenamed("lastName", "last_name") 
      .withColumnRenamed("secondLastName", "second_last_name") 
      .withColumnRenamed("dateOfBirth", "date_of_birth") 
      .withColumnRenamed("citizenId", "citizen_id") 
      .withColumnRenamed("createdAt", "created_at") 
      .withColumnRenamed("updatedAt", "updated_at") 


    // prepare the addresses data frame, but link it to Country, Currency, etc, while we"re at it.
    // in the process, we rename the long field names into short ones and drop unneeded columns
    
    addressesDF = addressesDF 
    .withColumnRenamed("id", "tempLocationId") 
    .withColumnRenamed("locations.val.addresses.val.id", "id") 
    .withColumnRenamed("locations.val.addresses.val.type", "type") 
    .withColumnRenamed("locations.val.addresses.val.zipCode", "zip_code") 
    .withColumnRenamed("locations.val.addresses.val.streetLine1", "street_line_1") 
    .withColumnRenamed("locations.val.addresses.val.streetLine2", "street_line_2") 
    .withColumnRenamed("locations.val.addresses.val.streetLine3", "street_line_3") 
    .withColumnRenamed("locations.val.addresses.val.apartmentNumber", "apartment_number") 
    .withColumnRenamed("locations.val.addresses.val.cityName", "city") 
    .withColumnRenamed("locations.val.addresses.val.countryCode", "country_code") 
    .withColumnRenamed("locations.val.addresses.val.stateCode", "state_code") 
    .drop("index")
    // now we will go through each table, renaming the long column names to short ones
    // (we drop the "index" columns too as we don"t need them)
    
    locationsDF = locationsDF 
    .withColumnRenamed("id", "tempCustomerAccountId") 
    .withColumnRenamed("locations.val.id", "id") 
    .withColumnRenamed("locations.val.name", "name") 
    .withColumnRenamed("locations.val.addresses", "addresses") 
    .drop("index")
    
    emailsDF = emailsDF 
    .withColumnRenamed("id","tempCustomerAccountId") 
    .withColumnRenamed("emails.val.id","id") 
    .withColumnRenamed("emails.val.value","value") 
    .withColumnRenamed("emails.val.primary","primary") 
    .drop("index")
    
    phoneNumbersDF = phoneNumbersDF 
    .withColumnRenamed("id","tempCustomerAccountId") 
    .withColumnRenamed("phoneNumbers.val.id","id") 
    .withColumnRenamed("phoneNumbers.val.number","number") 
    .withColumnRenamed("phoneNumbers.val.extension","extension") 
    .withColumnRenamed("phoneNumbers.val.region","region") 
    .withColumnRenamed("phoneNumbers.val.type","type") 
    .drop("index")
    
    // now we have all the tables we need and their relations. however, glue has used
    // auto-generated int ids to keep the relations. we want to use the ids inherent in the data.
    // also, for some relations, a separate table should be kept. thus, we will now use the 
    // int ids to do joins across tables, so we can take the actual ids that 
    // we want and create the tables we need. only after this will we drop the unwanted id fields. 
    // and we start with the lowest-level objects first. Country, Currency, State are done because 
    // we crafted those tables from scratch.
    
    // owner business account OR parent business account
    // here, we have a owner business account ID. however, we don"t know if it refers to a Business
    // Account or a Parent Business Account (which are stored in separate tables). So, we will now do 
    // a read to determine this, and then populate the correct relationship table.
      
    var baChildCustomerDF = customerAccountsDF.alias("c")
                       .join(
                             existingBusinessAccounts.alias("b"),
                             col("c.business_account_id") === col("b.id"),
                             "left")
                        .filter(col("b.id").isNotNull)
                        .select(col("b.id").as("business_account_id"), col("c.id").as("child_customer_id"))

    var pbaChildCustomerDF = customerAccountsDF.alias("c")
                       .join(
                             existingParentBusinessAccounts.alias("p"),
                             col("c.business_account_id") === col("p.id"),
                             "left")
                        .filter(col("p.id").isNotNull)
                        .select(col("p.id").as("parent_business_account_id"), col("c.id").as("child_customer_id"))

    //addresses and locations
    addressesDF = addressesDF.alias("a")
      .join( 
        locationsDF.alias("l"), 
        col("a.tempLocationId") === col("l.addresses"), 
        "left") 
      .select(col("a.*"), col("l.id").as("location_id")) 
      .drop("tempLocationId")
    
    //dont need locations.addresses anymore. drop it.
    locationsDF = locationsDF.drop("addresses")

    //customerAccount_location
    var customerAccountLocationsDF = locationsDF.alias("l") 
      .join(
        customerAccountsDF.alias("c"), 
        col("c.locations") === col("l.tempCustomerAccountId"), 
        "left") 
      .select(col("c.id").alias("customer_id"),col("l.id").alias("location_id"))
    
    //dont need locations.tempCustomerAccountId anymore. drop it.
    locationsDF = locationsDF.drop("tempCustomerAccountId")
    
    //customerAccount_email
    var customerAccountEmailsDF = emailsDF.alias("eml") 
      .join(
        customerAccountsDF.alias("c"), 
        col("c.emails") === col("eml.tempCustomerAccountId"), 
        "left") 
      .select(col("c.id").alias("customer_id"),col("eml.id").alias("email_id"))
    
    //dont need emailsDF.tempCustomerAccountId anymore. drop it.
    emailsDF = emailsDF.drop("tempCustomerAccountId")
    
    
    // customerAccount_phoneNumbers
    var customerAccountPhoneNumbersDF = phoneNumbersDF.alias("p") 
      .join(
        customerAccountsDF.alias("c"), 
        col("c.phone_numbers") === col("p.tempCustomerAccountId"),
        "left") 
      .select(col("c.id").alias("customer_id"),col("p.id").alias("phone_number_id"))
    
    //dont need phoneNumbersDF.tempCustomerAccountId anymore. drop it.
    phoneNumbersDF = phoneNumbersDF.drop("tempCustomerAccountId")

    // customerAccount_phoneNumbers
    var customerAccountPlaceOfBirthDF = placeOfBirthDF.alias("p") 
      .join(
        customerAccountsDF.alias("c"), 
        col("c.id") === col("p.tempCustomerAccountId"),
        "left") 
      .select(col("c.id").alias("customer_account_id"),col("p.id").alias("place_of_birth_id"))
    
    //dont need placeOfBirthDF.tempCustomerAccountId anymore. drop it.
    placeOfBirthDF = placeOfBirthDF.drop("tempCustomerAccountId")
    
    //all tables are made and now the only thing left is to delete the glue indices from the 
    //business accounts table as we don"t need them anymore

    customerAccountsDF = customerAccountsDF 
    .drop("phoneNumbers") 
    .drop("emails") 
    .drop("locations")
    println("ETL Job: Done Processing DataFrames")
    
    //need to drop duplicates from data frames
    
    addressesDF = addressesDF.dropDuplicates()
    locationsDF = locationsDF.dropDuplicates()
    phoneNumbersDF = phoneNumbersDF.dropDuplicates()
    emailsDF = emailsDF.dropDuplicates()
    placeOfBirthDF = placeOfBirthDF.dropDuplicates()
    customerAccountsDF = customerAccountsDF.dropDuplicates()
    customerAccountLocationsDF = customerAccountLocationsDF.dropDuplicates()
    customerAccountEmailsDF = customerAccountEmailsDF.dropDuplicates()
    customerAccountPhoneNumbersDF = customerAccountPhoneNumbersDF.dropDuplicates()
    customerAccountPlaceOfBirthDF = customerAccountPlaceOfBirthDF.dropDuplicates()
    
    
    //parentBA_childBADF = parentBA_childBADF.dropDuplicates()
    println("ETL Job: Done dropping duplicates")
    
    //now we need to take care of optional fields. "relationalize" doesn"t do a great
    //job of inferring the types of fields when they have no value (it"s because of the embedded Avro schema 
    //in the parquet file). And, if we get a single record with a missing field (optional), its type is 
    //kept as "null". This causes a problem when writing to JDBC. We need to explicitly set the types of 
    //the columns of all optional fields
    
    //furthermore, for all child elements, "relationalize" changes null or missing values
    //to empty strings. we want those nulls! again, the top level "businessAccounts" does
    //not suffer this problem. The nulls are intact there. we only need to do this for any
    //optional fields.
    
    if ( phoneNumbersDF.columns.contains("extension")) {
        phoneNumbersDF = phoneNumbersDF 
        .withColumn("extension", col("extension").cast(StringType)) 
        .withColumn("extension", when(col("extension") === "", null).otherwise(col("extension")))
    }

    if ( addressesDF.columns.contains("street_line_2")) {
        addressesDF = addressesDF 
        .withColumn("street_line_2", col("street_line_2").cast(StringType)) 
        .withColumn("street_line_2", when(col("street_line_2") === "", null).otherwise(col("street_line_2")))
    }

    if ( addressesDF.columns.contains("street_line_3")) {
        addressesDF = addressesDF 
        .withColumn("street_line_3", col("street_line_3").cast(StringType)) 
        .withColumn("street_line_3", when(col("street_line_3") === "", null).otherwise(col("street_line_3")))
    }

    if ( addressesDF.columns.contains("apartment_number")) {
        addressesDF = addressesDF 
        .withColumn("apartment_number", col("apartment_number").cast(StringType)) 
        .withColumn("apartment_number", when(col("apartment_number") === "", null).otherwise(col("apartment_number")))
    }
    
    //also need to cast date fields from string to date. they are strings because postgres wont accept them as they are naturally formatted.
    customerAccountsDF = customerAccountsDF
        .withColumn("created_at", col("created_at").cast(TimestampType)) 
        .withColumn("updated_at", col("updated_at").cast(TimestampType))
        .withColumn("date_of_birth", col("date_of_birth").cast(TimestampType)) 

    
    println("ETL Job: Done type casting optional parameters")

    //before we write each table to the db, we want to make sure we dont try to write any records that already exist.
    //it will cause duplication problems and/or constraint conflicts.
    //so, we have to join our tables with the existing tables and only keep records that are joined to "null" (record doesn"t exist)
    
    //make connection options per table, for reading DB
    
    //join new records with existing ones to only keep truly new records
    var newAddresses = getNewEntries(existingAddresses, addressesDF, addressesKeys)
    var newLocations = getNewEntries(existingLocations, locationsDF, locationsKeys)
    var newPhoneNumbers = getNewEntries(existingPhoneNumbers, phoneNumbersDF, phoneNumbersKeys)
    var newEmails = getNewEntries(existingEmails, emailsDF, emailsKeys)
    var newPlaceOfBirths = getNewEntries(existingPlaceOfBirths, placeOfBirthDF, placeOfBirthKeys)
    var newCustomerAccounts = getNewEntries(existingCustomerAccounts, customerAccountsDF, customerAccountsKeys)
    var newCustomerAccountsLocations = getNewEntries(existingCustomerAccountsLocations, customerAccountLocationsDF, customersLocationsKeys)
    var newCustomerAccountsPhoneNumbers = getNewEntries(existingCustomerAccountsPhoneNumbers, customerAccountPhoneNumbersDF, customersPhoneNumbersKeys)
    var newCustomerAccountsEmails = getNewEntries(existingCustomerAccountsEmails, customerAccountEmailsDF, customersEmailsKeys)
    var newCustomerAccountsPlaceOfBirths = getNewEntries(existingCustomerAccountsPlaceOfBirths, customerAccountPlaceOfBirthsDF, customersPlaceOfBirthsKeys)
    var newBAChildCustomer = getNewEntries(existingBusinessAccountChildCustomers, baChildCustomerDF, businessAccountChildCustomerKeys)
    var newPBAChildCustomer = getNewEntries(existingParentBusinessAccountChildCustomers, pbaChildCustomerDF, parentBusinessAccountChildCustomerKeys)
    
    // now we have all the tables we ultimately need (unique, new rows) as spark dataframes. convert them to glue dynamic frames for writing to DB
    
    var addressesDynF = DynamicFrame(newAddresses, glueContext)
    var locationsDynF = DynamicFrame(newLocations, glueContext)
    var phoneNumbersDynF = DynamicFrame(newPhoneNumbers, glueContext)
    var emailsDynF = DynamicFrame(newEmails, glueContext)
    var placeOfBirthsDynF = DynamicFrame(newPlaceOfBirths, glueContext)
    var customerAccountsDynF = DynamicFrame(newCustomerAccounts, glueContext)
    var customerAccountsLocationsDynF = DynamicFrame(newCustomerAccountsLocations, glueContext)
    var customerAccountsEmailsDynF = DynamicFrame(newCustomerAccountsEmails, glueContext)
    var customerAccountsPhoneNumbersDynF = DynamicFrame(newCustomerAccountsPhoneNumbers, glueContext)
    var customerAccountsPlaceOfBirthsDynF = DynamicFrame(newCustomerAccountsPlaceOfBirths, glueContext)
    var businessAccountChildCustomerDynF = DynamicFrame(newBAChildEmp, glueContext)
    var parentBusinessAccountChildCustomerDynF = DynamicFrame(newPBAChildEmp, glueContext)
    var customerOwnerBusinessAccountDynF = DynamicFrame(newEmpOwnerBA, glueContext)
    var customerOwnerParentBusinessAccountDynF = DynamicFrame(newEmpOwnerPBA, glueContext)
    println("ETL Job: Done converting Data Frames to Dynamic Frames")
    
    //finally, write dynamic frames to DB
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(locationConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(locationsDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(addressConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(addressesDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(phoneNumberConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(phoneNumbersDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(emailConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(emailsDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(placeOfBirthConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(placeOfBirthsDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(customerAccountConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(customerAccountsDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(customerAccountLocationConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(customerAccountsLocationsDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(customerAccountPhoneNumberConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(customerAccountsPhoneNumbersDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(customerAccountEmailConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(customerAccountsEmailsDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(customerAccountPlaceOfBirthConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(customerAccountsPlaceOfBirthsDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(businessAccountChildCustomerConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(businessAccountChildCustomerDynF)
    glueContext.getJDBCSink(catalogConnection = catalogueConnection, options = JsonOptions(parentBusinessAccountChildCustomerConnOptions), redshiftTmpDir = tempStorage).writeDynamicFrame(parentBusinessAccountChildCustomerDynF)
    println("ETL Job: Done writing to DB")
  
    Job.commit()  
  }
}