过滤 RDD 中的空分区答案

【问题标题】：Filtering empty partitions in RDD过滤 RDD 中的空分区
【发布时间】：2020-09-17 00:24:44
【问题描述】：

有没有办法过滤 RDD 中的空分区？分区后我有一些空分区，我不能在操作方法中使用它们。

我在 Scala 中使用 Apache Spark

【问题讨论】：

为什么不能使用分区。你能展示整个代码和工作流程吗？什么样的操作只考虑分区？
用完整的例子检查我的答案。根据示例，只能存在非空分区。现在您的操作无法按照您的要求应用于空分区
@UninformedUser 我正在处理运行其他工具的拆分文件。我不能使用空分区，因为工具会抛出异常：工具从标准输入获取数据并且它在为空时抛出异常
@Patrice333 恕我直言，这似乎更像是那些工具的问题，而不是你的问题，应该在这些工具中解决。

标签： scala apache-spark filtering rdd partition

【解决方案1】：

这是我的示例数据

  val sc = spark.sparkContext
  val myDataFrame = spark.range(20).toDF("mycol").repartition($"mycol")
  myDataFrame.show(false)

输出：

+-----+
|mycol|
+-----+
|19   |
|0    |
|7    |
|6    |
|9    |
|17   |
|5    |
|1    |
|10   |
|3    |
|12   |
|8    |
|11   |
|2    |
|4    |
|13   |
|18   |
|14   |
|15   |
|16   |
+-----+

在上面的代码中，当您对列进行重新分区时，将创建 200 个分区，因为 spark.sql.shuffle.partitions = 200 其中许多未使用或空分区，因为数据只有 10 个数字（我们试图将 20 个数字放入 200 个分区意味着....大部分分区都是空的.... :-))

1) 准备一个长累加器变量来快速计算非空分区。
2) 将所有非空分区添加到累加器变量中，如下例所示。

 val nonEmptyPartitions = sc.longAccumulator("nonEmptyPartitions")
 myDataFrame.foreachPartition(partition =>
    if (partition.length > 0) nonEmptyPartitions.add(1))

删除非空分区（意味着合并它们...less shuffle/ minimum shuffle）。
打印出来。


val finalDf = myDataFrame.coalesce(nonEmptyPartitions.value.toInt)
println(s"nonEmptyPart : ${nonEmptyPartitions.value.toInt}")
println(s"df.rdd.partitions.length :  ${myDataFrame.rdd.getNumPartitions}")
println(s"finalDf.rdd.partitions.length  :  ${finalDf.rdd.getNumPartitions}")

打印它们...

结果：

nonEmptyPart : 20
df.rdd.partitions.length :  200
finalDf.rdd.partitions.length  :  20

证明所有非空分区都被删除...

  myDataFrame.withColumn("partitionId", org.apache.spark.sql.functions.spark_partition_id)
.groupBy("partitionId")
.count
.show

结果打印分区明智记录计数：

 +-----------+-----+
|partitionId|count|
+-----------+-----+
|128        |1    |
|190        |1    |
|140        |1    |
|164        |1    |
|5          |1    |
|154        |1    |
|112        |1    |
|107        |1    |
|4          |1    |
|49         |1    |
|69         |1    |
|77         |1    |
|45         |1    |
|121        |1    |
|143        |1    |
|58         |1    |
|11         |1    |
|150        |1    |
|68         |1    |
|116        |1    |
+-----------+-----+

注意：

使用 spark_partition_id 仅用于演示/调试目的，不用于生产目的。
我将 200 个分区（由于 column 上的重新分区）减少到 20 个非空分区。

结论：

最后你摆脱了多余的没有任何数据的空分区，并避免了不必要的安排在空分区上的虚拟任务。

【讨论】：

如果有用，你想accept the answer & vote吗？
是的，这是我使用的解决方案。但是我的分区是拆分文件，因此合并方法会大大增加处理时间。
如果您想为您的操作忽略空分区以及任务的不必要工作。你有/应该做合并或重新分区（非空分区）

【解决方案2】：

根据您提供的少量信息，我可以考虑两种选择。使用 mapPartitions 并捕获空迭代器并返回它们，同时处理非空迭代器。

rdd.mapPartitions { case iter => if(iter.isEmpty) { iter } else { ??? } }

或者你可以使用repartition，去掉空分区。

rdd.repartition(10) // or any proper number

【讨论】：

我必须在分区 rdd 上工作，而不是分区，所以第一个选项对我不利。但第二 - 它增加了很多处理时间。我用了colesce()，但是对我不好

【解决方案3】：

如果您不知道列中的不同值，并且希望避免出现空分区，您可以使用countApproxDistinct() 作为：

df.repartition(df.rdd.countApproxDistinct().toInt)

如果您希望过滤现有的空分区并重新分区，您可以使用 Sasa 建议的解决方案

或者：

df.repartition(df.mapPartitions(part => List(part.length).iterator).collect().count(_ != 0)).df.getNumPartitions)

但是，在以后的情况下，分区可能包含也可能不包含按值记录的记录。

【讨论】：