如何（同样）在火花数据帧中分割数组数据

Question

我有以下形式的数据框：

import scala.util.Random
val localData = (1 to 100).map(i => (i,Seq.fill(Math.abs(Random.nextGaussian()*100).toInt)(Random.nextDouble)))
val df = sc.parallelize(localData).toDF("id","data")

|-- id: integer (nullable = false)
|-- data: array (nullable = true)
|    |-- element: double (containsNull = false)


df.withColumn("data_size",size($"data")).show

+---+--------------------+---------+
| id|                data|data_size|
+---+--------------------+---------+
|  1|[0.77845301260182...|      217|
|  2|[0.28806915178410...|      202|
|  3|[0.76304121847720...|      165|
|  4|[0.57955190088558...|        9|
|  5|[0.82134215959459...|       11|
|  6|[0.42193739241567...|       57|
|  7|[0.76381645621403...|        4|
|  8|[0.56507523859466...|       93|
|  9|[0.83541853717244...|      107|
| 10|[0.77955626749231...|      111|
| 11|[0.83721643562080...|      223|
| 12|[0.30546029947285...|      116|
| 13|[0.02705462199952...|       46|
| 14|[0.46646815407673...|       41|
| 15|[0.66312488908446...|       16|
| 16|[0.72644646115640...|      166|
| 17|[0.32210572380128...|      197|
| 18|[0.66680355567329...|       61|
| 19|[0.87055594653295...|       55|
| 20|[0.96600507545438...|       89|
+---+--------------------+---------+

现在我想应用一个昂贵的 UDF，计算时间与数据数组的大小成正比。我想知道如何重新分区我的数据，以便每个分区具有大致相同数量的“记录*数据大小”（即，数据点不仅仅是记录）。

如果只是这样做df.repartition(100)，我可能会得到一些包含一些非常大的数组的分区，这些分区是整个火花阶段的瓶颈（所有其他任务已经完成）。当然，如果我可以选择疯狂数量的分区，这将（几乎）确保每个记录都在一个单独的分区中。但是还有别的方法吗？

Answer 1

正如您所说，您可以增加分区数量。我通常使用内核数的倍数：spark context default parallelism * 2-3 ..
在您的情况下，您可以使用更大的乘数。

另一种解决方案是以这种方式过滤拆分您的 df：

• df 只有更大的数组
• df 和其余的

然后，您可以重新分区它们中的每一个，执行计算并将它们合并回来。

请注意，重新分区可能会很昂贵，因为您有大行要随机播放。

你可以看看这些幻灯片（27+）：https : //www.slideshare.net/SparkSummit/custom-applications-with-sparks-rdd-spark-summit-east-talk-by-tejas-patil

他们遇到了非常糟糕的数据倾斜，不得不以一种有趣的方式处理它。