Notes

String partition="";

//Choose partition cols based on Load_Type_Code

//System.out.println("Load_Type_Code = " + row1.Load_Type_Code);
if ((row1.Load_Type_Code.equals("FLUSH_N_FILL")) || (row1.Load_Type_Code.equals("flush_n_fill")))
partition = context.Master_Hive_Partition_Cols;

else //if ((row1.Load_Type_Code.equals("INSERT_ONLY")) || (row1.Load_Type_Code.equals("insert_only")))
partition = context.Transaction_Hive_Partition_Cols;

String Query3 = input_row.Query3.replace(",,",",");

globalMap.put("Query3", Query3);

//System.out.println("Query3" + (String)globalMap.get("Query3"));



globalMap.put("partition", partition);

// CREATE the database if not present

String create_err_db = "CREATE DATABASE IF NOT EXISTS " + context.Hive_Err_DBName;
String create_tmp_db = "CREATE DATABASE IF NOT EXISTS " + context.Hive_Tmp_DBName;
String create_tgt_db = "CREATE DATABASE IF NOT EXISTS " + context.Hive_Tgt_DBName;


//DROP and CREATE the HZ table

String tgt_drop_query = "DROP TABLE IF EXISTS " + context.Hive_Tgt_DBName + "." + input_row.Data_Lake_Object;

String tgt_query = "CREATE EXTERNAL TABLE IF NOT EXISTS " + context.Hive_Tgt_DBName + "." + input_row.Data_Lake_Object + " (" + (String)globalMap.get("Query3") + context.Hive_ABC_Cols + ") PARTITIONED BY (" + (String)globalMap.get("partition") + ") STORED AS ORC LOCATION '" + context.ADLS_Name + row1.Target_File_Location + "'";

//DROP and CREATE the ERR table

String err_drop_query = "DROP TABLE IF EXISTS " + context.Hive_Err_DBName + "." + row1.Error_Object;


String err_query = "CREATE EXTERNAL TABLE IF NOT EXISTS " + context.Hive_Err_DBName + "." + row1.Error_Object + " (" + input_row.Query2 + ",err_msg string,ins_gmt_ts bigint,ins_usr_id string,src_sys_id bigint,load_job_nr bigint) ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe' WITH SERDEPROPERTIES ('field.delim' = '|~|') STORED AS TEXTFILE LOCATION '" + context.ADLS_Name + row1.Rejected_Record_Location + "'";

//DROP and CREATE the TMP table

String tmp_drop_query = "DROP TABLE IF EXISTS " + context.Hive_Tmp_DBName + "." + row1.Target_Temp_Object;

String tmp_query = "CREATE EXTERNAL TABLE IF NOT EXISTS " + context.Hive_Tmp_DBName + "." + row1.Target_Temp_Object + " (" + input_row.Query1 + ") ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.MultiDelimitSerDe' WITH SERDEPROPERTIES ('field.delim' = '|~|') STORED AS TEXTFILE LOCATION '" + context.ADLS_Name + row1.Temporary_Location + "'";

//Assign to global variables

globalMap.put("err_query", err_query);
globalMap.put("err_drop_query", err_drop_query);
globalMap.put("tmp_query", tmp_query);
globalMap.put("tmp_drop_query", tmp_drop_query);
globalMap.put("tgt_query", tgt_query);
globalMap.put("tgt_drop_query", tgt_drop_query);
globalMap.put("create_err_db", create_err_db);
globalMap.put("create_tmp_db", create_tmp_db);
globalMap.put("create_tgt_db", create_tgt_db);

System.out.println("/**** Hive tables creation started for '" + row1.Target_Data_Lake_Object + "' with Load_Type_Code = " + row1.Load_Type_Code + "****/");
System.out.println("tmp query = " + (String)globalMap.get("tmp_query"));
System.out.println(" ");
System.out.println("err query = " + (String)globalMap.get("err_query"));
System.out.println(" ");
System.out.println("tgt query = " + (String)globalMap.get("tgt_query"));