Die Verwendung von dropDuplicates im Datenframe verursacht Änderungen in der Partitionsnummer

Question

Ich habe einen großen Datenrahmen, den ich mit 800 Partitionen erstellt habe.

df.rdd.getNumPartitions() 
800 

Als ich dropDuplicates auf dem Datenrahmen verwenden, ändert es die Partitionen für mich 200

df = df.dropDuplicates() 
df.rdd.getNumPartitions() 
200 

Dieses Verhalten führt Problem auf dem Standard, wie es aus der Erinnerung führen wird.

Haben Sie Vorschläge zur Behebung dieses Problems? Ich habe versucht, spark.sql.shuffle.partition auf 800 zu setzen, aber es funktioniert nicht. Dank

Answer 1

Dies passiert, weil dropDuplicates einen Shuffle erfordert. Wenn Sie eine bestimmte Anzahl von Partitionen erhalten möchten, sollten Sie spark.sql.shuffle.partitions gesetzt (der Standardwert ist 200)

df = sc.parallelize([("a", 1)]).toDF() 
df.rdd.getNumPartitions() 
## 8 

df.dropDuplicates().rdd.getNumPartitions() 
## 200 

sqlContext.setConf("spark.sql.shuffle.partitions", "800") 

df.dropDuplicates().rdd.getNumPartitions() 
## 800 

Ein alternativer Ansatz (Spark-1.6+) zunächst zu partitionieren ist:

df.repartition(801, *df.columns).dropDuplicates().rdd.getNumPartitions() 
## 801 

Es ist etwas flexibler, aber weniger effizient, da keine lokale Aggregation durchgeführt wird.

Answer 2

fand ich die Lösung bei Removing duplicates from rows based on specific columns in an RDD/Spark DataFrame

Verwenden reduceByKey statt dropDuplicates. reduceByKey hat auch eine Option, die Anzahl der Partitionen für die endgültige rdd anzugeben.

Der Nachteil der Verwendung von reduceByKey in diesem Fall ist es ist langsam.