Ich habe zwei RDDs. Eine RDD liegt zwischen 5 und 10 Millionen Einträgen und die andere RDD zwischen 500 und 750 Millionen Einträgen. Irgendwann muss ich diese beiden RDDS mit einem gemeinsamen Schlüssel verbinden.

val rddA = someData.rdd.map { x => (x.key, x); } // 10-million
val rddB = someData.rdd.map { y => (y.key, y); } // 600-million
var joinRDD = rddA.join(rddB);

Wenn sich spark für diesen Join entscheidet, entscheidet er sich für einen ShuffledHashJoin. Dadurch werden viele der Elemente in rddB im Netzwerk gemischt. Ebenso werden einige rddA-Dateien im Netzwerk gemischt. In diesem Fall ist rddA zu "groß", um als Broadcast-Variable verwendet zu werden, aber es scheint, als wäre ein BroadcastHashJoin effizienter. Gibt es einen Hinweis, um einen BroadcastHashJoin zu verwenden? (Apache Flink unterstützt dies durch Join-Hinweise.)

Wenn nicht, ist dies die einzige Möglichkeit, den autoBroadcastJoinThreshold zu erhöhen?

Update 7/14

Mein Leistungsproblem scheint in der Neupartitionierung begründet zu sein. Normalerweise würde ein RDD-Lesevorgang von HDFS blockweise partitioniert, aber in diesem Fall war die Quelle eine Parkettdatenquelle [die ich erstellt habe]. Wenn spark (Databricks) die Parkettdatei schreibt, schreibt es eine Datei pro Partition und liest identisch eine Partition pro Datei. Die beste Antwort, die ich gefunden habe, ist, dass Sie während der Produktion der Datenquelle die Parkettwanne (die dann natürlich mit partitioniert wird) ausschreiben und diese als rddB verwenden, um sie dann nach Schlüsseln zu partitionieren.

Die Antwort ist richtig, aber ich denke, die Details zur Parkettdatenquelle können für andere nützlich sein.