Le partitionnement Spark pour l'écriture de fichier est très lent

Question

Lors de l'écriture d'un fichier sur HDFS à l'aide de Spark, cette opération est assez rapide lorsque vous n'utilisez pas le partitionnement. Au lieu de cela, lorsque j'utilise le partitionnement pour écrire le fichier, le délai d'écriture augmente de facteur ~ 24.

Pour le même fichier, écrire sans partition prend environ 600ms. Ecrire avec partition par Id (va générer exactement 1.000 partitions, comme il y a 1.000 ids dans le fichier) ça prend environ 14 secondes.

Certains d'entre vous ont-ils la même expérience que l'écriture d'un fichier partitionné prend beaucoup de temps? Quelle est la cause de cette situation, peut-être que Spark doit créer 1.000 dossiers et fichiers pour chaque partition? Savez-vous comment cela peut être accéléré?

val myRdd = streamedRdd.map { case ((id, metric, time), value) => Record(id, metric, getEpoch(time), time, value) } 

val df = myRdd.toDF 

df.write.mode(SaveMode.Append) 
.partitionBy("id") 
.parquet(path) 

Answer 1

exécuteurs Spark communiquer avec HDFS pour écrire les données qu'ils ont, cela dépend de la façon dont vos données sont réparties sur le cluster après le partitionnement.

Apparemment, pour de plus petits morceaux de données, le temps nécessaire pour établir les connexions de plusieurs nœuds exécuteurs à HDFS et écrire serait plus important si on le compare à l'écriture séquentielle du fichier entier.

Comment éviter ceci:

par étincelle par défaut partitionne les données à l'aide partitionneur Hash (tables de hachage de la clé et la clé avec la même hachage va au même noeud) essayez de spécifier le partitionneur Range, s'il vous plaît trouver les extraits d'échantillons ci-dessous:

L'extrait suivant utilise Hash partitionneur yourRdd.groupByKey() saveAsTextFile ("HDFS PATH");.

L'extrait suivant utilise notre gamme personnalisée partitionneur Il crée 8 partitions comme mentionné dans RangePartitioner(8, yourRdd) et de l'écriture à travers 8 connexions serait un meilleur choix alors écrit, par 1000 connexions.

val tunedPartitioner = new RangePartitioner(8, yourRdd) 
val partitioned = yourRdd.partitionBy(tunedPartitioner).saveAsTextFile("HDFS PATH"); 

Encore une fois c'est un compromis entre les données à écrire et le nombre de partitions que vous créez.