将rdd转换为数据帧时的Spark重新分区性能

Question

我是scala/sark世界的新手，最近开始了一项任务，它读取一些数据，处理数据并将其保存在S3上。我阅读了一些关于stackoverflow的主题/问题，这些主题/问题涉及重分区/合并性能和最佳分区数（如本例）。假设我有正确的分区数，我的问题是，在将rdd转换为数据帧时，对它进行重新分区是个好主意吗？下面是我的代码目前的样子：

val dataRdd = dataDf.rdd.repartition(partitions)
      .map(x => ThreadedConcurrentContext.executeAsync(myFunction(x)))
      .mapPartitions( it => ThreadedConcurrentContext.awaitSliding(it = it, batchSize = asyncThreadsPerTask, timeout = Duration(3600000, "millis")))

val finalDf = dataRdd
      .filter(tpl => tpl._3 != "ERROR")
      .toDF()

这是我打算做的（过滤后重新分区数据）：

val finalDf = dataRdd
          .filter(tpl => tpl._3 != "ERROR")
          .repartition(partitions)
          .toDF()

我的问题是，这样做是个好主意吗？这里有性能提升吗？

注1：过滤器通常会删除10%的原始数据。

注2：这是我用来运行上述代码的spark-submit命令的第一部分：

spark-submit --master yarn --deploy-mode client --num-executors 4 --executor-cores 4 --executor-memory 2G --driver-cores 4 --driver-memory 2G

Answer 1

问题的答案取决于dataRdd的大小、分区数、执行器核以及HDFS集群的处理能力。

考虑到这一点，您应该使用分区的不同值在集群上运行一些测试，并完全删除repartition，以便对其进行微调并找到准确的答案。

例如-如果您指定分区=8和执行器-核心=4，那么您将充分利用所有核心，但是如果您的dataRdd的大小只有1GB，那么重新分区没有任何优势，因为它会触发shuffle，从而影响性能。此外，如果您的HDFS集群的处理能力较低或负载较重，那么会因此产生额外的性能开销。

如果您的HDFS集群上有足够的可用资源，并且您有一个大的(比如超过100GB) dataRDD，那么< code >重新分区应该有助于提高上面示例中配置值的性能。