Speichern Sie Spark-Datenframe als dynamische partitionierte Tabelle in Hive

Question

Ich habe eine Beispielanwendung arbeiten zum Lesen von CSV-Dateien in einen Datenrahmen. Der Datenrahmen kann mit der Methode df.saveAsTable(tablename,mode) in einer Hive-Tabelle im Parkettformat gespeichert werden.

Der obige Code funktioniert gut, aber ich habe so viele Daten für jeden Tag, dass ich die Hive-Tabelle basierend auf dem creationdate (Spalte in der Tabelle) dynamisch partitionieren möchte.

gibt es eine Möglichkeit, den Datenrahmen dynamisch zu partitionieren und ihn im Stock-Warehouse zu speichern. Unterlassen Sie die Hard-Codierung der INSERT-Anweisung unter Verwendung von hivesqlcontext.sql(insert into table partittioin by(date)....). How to save DataFrame directly to Hive?

jede Hilfe ist sehr willkommen:

Frage kann als Erweiterung zu berücksichtigen.

Answer 1

Ich glaube, es so etwas wie dies funktioniert:

df ist ein Datenrahmen mit Jahr, Monat und anderen Spalten

df.write.partitionBy('year', 'month').saveAsTable(...) 

oder

df.write.partitionBy('year', 'month').insertInto(...) 

Answer 2

konnte ich partitioniert hive schreiben Tabelle mit df.write().mode(SaveMode.Append).partitionBy("colname").saveAsTable("Table")

Ich musste die Fo aktivieren Eigenschaften, um es zum Laufen zu bringen.

 
hiveContext.setConf("hive.exec.dynamic.partition", "true") 
hiveContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict") 

Answer 3

Ich sah auch die gleiche Sache, aber mit folgenden Tricks, die ich gelöst.

1. Wenn wir eine Tabelle als partitioniert und dann partitionierte Spalte Groß-und Kleinschreibung beachten.

2. Partitionierte Spalte sollte in DataFrame mit demselben Namen vorhanden sein (Groß-/Kleinschreibung beachten). Code:

   var dbName="your database name" 
   var finaltable="your table name" 
   
   // First check if table is available or not.. 
   if (sparkSession.sql("show tables in " + dbName).filter("tableName='" +finaltable + "'").collect().length == 0) { 
       //If table is not available then it will create for you.. 
       println("Table Not Present \n Creating table " + finaltable) 
       sparkSession.sql("use Database_Name") 
       sparkSession.sql("SET hive.exec.dynamic.partition = true") 
       sparkSession.sql("SET hive.exec.dynamic.partition.mode = nonstrict ") 
       sparkSession.sql("SET hive.exec.max.dynamic.partitions.pernode = 400") 
       sparkSession.sql("create table " + dbName +"." + finaltable + "(EMP_ID  string,EMP_Name   string,EMP_Address    string,EMP_Salary bigint) PARTITIONED BY (EMP_DEP STRING)") 
       //Table is created now insert the DataFrame in append Mode 
       df.write.mode(SaveMode.Append).insertInto(empDB + "." + finaltable) 
   }