Zapisz ramkę danych Spark jako dynamicznie podzieloną tabelę w Hive

Question

Mam przykładową aplikację do odczytu z plików CSV do ramki danych. Ramka danych może być przechowywana w tabeli Hive w formacie parkietu, przy użyciu metody: .

Powyższy kod działa poprawnie, ale mam tyle danych na każdy dzień, że chcę dynamicznie partycjonować tabelę ula na podstawie daty utworzenia (kolumna w tabeli).

czy istnieje sposób na dynamiczną partycję ramki danych i zapisanie jej w magazynie magazynu. Chcesz powstrzymać się od zakodowania instrukcji wstawiania za pomocą hivesqlcontext.sql(insert into table partittioin by(date)....).

Pytanie można uznać za rozszerzenie: How to save DataFrame directly to Hive?

każda pomoc jest mile widziane.

Answer 1

wierzę, że działa coś takiego:

df jest dataframe z rok, miesiąc i innych kolumn

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

lub

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

Answer 2

udało mi się napisać do podzielono ula tabeli przy użyciu df.write().mode(SaveMode.Append).partitionBy("colname").saveAsTable("Table")

Musiałem włączyć fo dzięki właściwościom, które sprawiają, że działa.

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

Answer 3

Zmierzyłem się również z tą samą sztuczką, ale rozwiązałem ją.

1. Kiedy robimy tabelę jako podzieloną na partycje, a następnie partycjonowana kolumna staje się wrażliwa na wielkość liter.

2. Partycjonowana kolumna powinna być obecna w DataFrame o tej samej nazwie (wielkość liter ma znaczenie). Kod:

   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) 
   }