如何参数化将数据帧写入配置单元表

Question

我在RBDMS中有一个表列表(跨不同类别(，我想提取并保存在配置单元中，我想参数化，以便能够将类别名称附加到配置单元中的输出位置。例如，我有一个类别"；"雇员"；，我希望能够将从RDBMS中提取的表以"；hive_db.emplloyee_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

它没有将提取的数据帧写入hive，而是只打印列名

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

如何将DF写入配置单元表？

Answer 1

你能试试这种方法吗，把你的桶图改成这样(我已经为t1做了，请为t2&t3做同样的事情(，

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

并用足够的论证替换CCD_ 1，如(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")