Merge große Anzahl von Spark-Datenframes in eine

Question

Ich Frage eine zwischengespeicherte Hive-Temp-Tabelle mit verschiedenen Abfragen, die verschiedenen Bedingungen mehr als 1500 Mal innerhalb einer for-Schleife. Ich muss sie alle mit unionAll innerhalb der Schleife zusammenführen. Aber ich bekomme Stackoverflow-Fehler aufgrund der Tatsache, dass Spark nicht mit der RDD-Linie Schritt halten kann.

Pseudocode:

df=[from a hive table] 
tableA=[from a hive table] 
tableA.registerTempTable("tableA") 
HiveContext.sql('CACHE TABLE tableA') 

for i in range(0,2000): 
    if (list[0]['column1']=='xyz'): 
     df1=query something from tableA 
     df=df.unionAll(df1) 
    elif(): 
     df1=query something from tableA 
     df=df.unionAll(df1) 
    elif(): 
     df1=query something from tableA 
     df=df.unionAll(df1) 
    elif(): 
     df1=query something from tableA 
     df=df.unionAll(df1) 
    else: 
     df1=query something from tableA 
     df=df.unionAll(df1) 

Dieser Stackoverflow-Fehler aufgrund RDD Linie hart zu werden wirft. So versuchte ich Checkpointing wie folgt:

for i in range(0,2000): 
    if (list[0]['column1']=='xyz'): 
     df1=query something from tableA 
     df=df.unionAll(df1) 
    elif(): 
     df1=query something from tableA 
     df=df.unionAll(df1) 
    else: 
     df1=query something from tableA 
     df=df.unionAll(df1) 
    df.rdd.checkpoint 
    df = sqlContext.createDataFrame(df.rdd, df.schema) 

Ich habe den gleichen Fehler. Also habe ich SaveAsTable ausprobiert, was ich immer vermeiden wollte, weil die Jobübergabe zwischen den einzelnen HQL-Abfragen und dem Hive innerhalb einer Schleife verzögert war. Aber dieser Ansatz hat gut funktioniert.

Ich brauche Hilfe bei der Vermeidung von Speichern des Datenrahmens in Bienenstock innerhalb der Schleife. Ich möchte die dfs in einer Art und Weise zusammenführen, die In-Memory und effizient ist. Eine der anderen Optionen, die ich ausprobiert habe, ist, das Abfrageergebnis direkt in eine temporäre Tabelle einzufügen, für die ich einen Fehler erhalte: kann nicht in eine RDD-basierte Tabelle eingefügt werden.

Answer 1

vielleicht, Tempentabelle für Ergebnis wird funktionieren.

df1="query something from tableA".registerTempTable("result") 
sqlContext.sql("Insert into result query something from tableA")