Как параметризовать запись фрейма данных в таблицу улья - PullRequest
Новая
       13

Как параметризовать запись фрейма данных в таблицу улья

0 голосов
/ 04 августа 2020

У меня есть список таблиц (по разным категориям) в RBDMS, которые я хочу извлечь и сохранить в улье, и я хочу параметризовать таким образом, чтобы я мог прикрепить имя категории к месту вывода в улей. Например, у меня есть категория «сотрудник», я хочу иметь возможность сохранить извлеченную таблицу из СУБД в формате «hive_db.employee_some_other_random_name»

У меня есть код, как показано ниже 

    val category = "employee"
    val tableList = List("schema.table_1", "schema.table_2", "schema.table_3")

    val tableMap = Map("schema.table_1" -> "table_1",
    "schema.table_2" -> "table_2",
    "schema.table_3" -> "table_3")

    val queryMap = Map("table_1" -> (select * from table_1) tble,
    "table_2" -> (select * from table_2) tble,
    "table_3" -> (select * from table_3) tble)

    val tableBucketMap = Map("table_1" -> "bucketBy(80,\"EMPLOY_ID\",\"EMPLOYE_ST\").sortBy(\"EMPLOY_ST\").format(\"parquet\")",
    "table_2" -> "bucketBy(80, \"EMPLOY_ID\").sortBy(\"EMPLOY_ID\").format(\"parquet\")",
    "table_3" -> "bucketBy(80, \"EMPLOY_ID\", \"SAL_ID\", \"DEPTS_ID\").sortBy(\"EMPLOY_ID\").format(\"parquet\")")

     for (table <- tableList){
       val tableName = tableMap(table)
       val print_start = "STARTING THE EXTRACTION PROCESSING FOR TABLE: %s"
       val print_statement = print_start.format(tableName)
       println(print_statement)

       val extract_query = queryMap(table)
       val query_statement_non = "Query to extract table %s is: "
       val query_statement = query_statement_non.format(tableName)
       println(query_statement + extract_query)


       val extracted_table = spark.read.format("jdbc")
         .option("url", jdbcURL)
         .option("driver", driver_type)
         .option("dbtable", extract_query)
         .option("user", username)
         .option("password", password)
         .option("fetchsize", "20000")
         .option("queryTimeout", "0")
         .load()

       extracted_table.show(5, false)
       //saving extracted table in hive
       val tableBucket = tableBucketMap(table)
       val output_loc = "hive_db.%s_table_extracted_for_%s"
       val hive_location = output_loc.format(category, tableName)
       println(hive_location)

       val saving_table = "%s.write.%s.saveAsTable(\"%s\")"
       saving_table.format(extracted_table, tableBucket, hive_location)
       println(saving_table.format(extracted_table, tableBucket, hive_location))
  
       val print_end = "COMPLETED EXTRACTION PROCESS FOR TABLE: %s"
       val print_end_statement = print_end.format(tableName)
       println(print_end_statement)

У меня есть результат для первой таблицы. Тот же результат применим и к другим таблицам .. 

STARTING THE EXTRACTION PROCESSING FOR TABLE: table_1
Query to extract table table_1 is: (select * from table_1) tble
+---------+--------------------+
|EMPLOY_ID|EMPLOYE_NM          |
+---------+--------------------+
|1        |WELLINGTON          |
|2        |SMITH               |
|3        |CURLEY              |
|4        |PENDRAGON           |
|5        |KEESLER             |
+---------+--------------------+
only showing top 5 rows

hive_db.employee_table_extracted_for_table_1

[EMPLOY_ID: int, EMPLOYE_NM: string].write.bucketBy(80, "EMPLOY_ID", "EMPLOYE_NO").sortBy("EMPLOY_ID").format("parquet").saveAsTable("hive_db.employee_table_extracted_for_table_1")

COMPLETED EXTRACTION PROCESS FOR TABLE: table_1

Вместо записи извлеченного фрейма данных в куст, он просто напечатал имена столбцов 

[EMPLOY_ID: int, EMPLOYE_NM: String].write............saveAsTable("hive_db.employee_table_extracted_for_table_1")

Как я могу заставить его писать DF в таблицу-улей?

1 Ответ

1 голос
/ 04 августа 2020

можете ли вы попробовать этот подход, измените карту сегментов следующим образом (я сделал для t1, сделайте то же самое для t2 и t3), 

    val tableBucketMap = Map("table_1" -> "80,\"employe_st\"")

и замените df.bucketBy() на достаточно аргументы типа (numBuckets: Int, colName: String, colNames: String*)

   
           val stringArr=tableBucket.split(",")
           val numBuckets=stringArr(0).toInt
           val colName=stringArr(1)
           
           extracted_table.write.mode("append").bucketBy(numBuckets,colName).format("parquet").saveAsTable(hive_location)

этот подход решит упомянутую проблему 

[EMPLOY_ID: int, EMPLOYE_NM: String].write............saveAsTable("hive_db.employee_table_extracted_for_table_1")
...