Wie wird der Spark-Dataframe an jedem Executor-Knoten gesammelt?

Question

Meine Anwendung liest eine große Parkettdatei und führt einige Datenextraktionen durch, um zu einem kleinen Funken-Datenframeobjekt zu gelangen. Der gesamte Inhalt dieses Datenrahmens muss bei jedem Executor-Knoten für die nächste Phase der Berechnung vorhanden sein. Ich weiß, dass ich diese von collect-Sendung tun kann, wie in diesem pyspark Schnipsel

sc = pyspark.SparkContext() 
sqlc = HiveContext(sc) 

# --- register hive tables and generate spark dataframe 
spark_df = sqlc.sql('sql statement') 

# collect spark dataframe contents into a Pandas dataframe at the driver 
global_df = spark_df.toPandas() 

# broadcast Pandas dataframe to all the executor nodes 
sc.broadcast(global_df) 

Ich frage mich: Gibt es eine effizientere Methode, dies zu tun? Es scheint, dass dieses Muster den Treiberknoten zu einem Engpass macht.

Answer 1

Es hängt davon ab, was Sie mit Ihrem kleinen Datenrahmen tun müssen. Wenn Sie mit einem großen verbinden müssen, kann Spark solche Joins optimieren, die kleine Datenframes automatisch übertragen. Die maximale Größe des Datenrahmens, der gesendet werden kann, wird durch die Option spark.sql.autoBroadcastJoinThreshold konfiguriert, wie in der Dokumentation http://spark.apache.org/docs/latest/sql-programming-guide.html#other-configuration-options

beschrieben