在HDFS上预共分组表和在Spark中用零洗牌读取表

Question

作为spark作业的一部分，我有两个表要加入/合并，这在每次运行作业时都会引起很大的洗牌。我想通过存储一次共分组数据来摊销所有工作的成本，并将已经共分组的数据作为定期Spark运行的一部分，以避免洗牌。

为了实现这一点，我将一些HDFS中的数据存储在拼花板格式中。我正在使用Parquet重复字段来实现以下模式

（日期，[aRecords]，[bRecords])

在这种情况下，aRecords和bRecords似乎可以有效地按日期合并。我可以执行如下操作：

case class CogroupedData(date: Date, aRecords: Array[Int], bRecords: Array[Int])

val cogroupedData = spark.read.parquet("path/to/data").as[CogroupedData]

//Dataset[(Date,Int)] where the Int in the two sides multiplied
val results = cogroupedData
    .flatMap(el => el.aRecords.zip(el.bRecords).map(pair => (el.date, pair._1 * pair._2)))

两者之间的区别在于，我避免了对已经合并的数据进行洗牌，合并的成本通过在HDFS上持久化来摊销。

现在回答问题。从cogrouped dataset中，我希望派生两个分组数据集，这样我就可以使用标准的Spark SQL运算符（如cogroup、join等）而不会导致洗牌。这似乎是可能的，因为第一个代码示例工作，但Spark仍然坚持在我加入/groupbykey/cogroup等时对数据进行散列/洗牌。

以下面的代码示例为例。我希望有一种方法，我们可以在执行联接时运行下面的内容而不会引起洗牌。

val cogroupedData = spark.read.parquet("path/to/data").as[CogroupedData]

val aRecords = cogroupedData
    .flatMap(cog => cog.aRecords.map(a => (cog.date,a)))
val bRecords = cogroupedData
    .flatMap(cog => cog.bRecords.map(b => (cog.date,b)))

val joined = aRecords.join(bRecords,Seq("date"))

查看文献，如果cogroupedData有一个已知的分区器，那么接下来的操作不应该引起洗牌，因为它们可以使用RDD已经分区的事实并保留分区器。

我认为要实现这一点，需要使用已知的分区器获得一个cogroupedData DataSet/RDD，而不会导致洗牌。

null

有人有什么想法吗？

Answer 1

你在这里犯了两个错误。

>

今天(Spark2.3)，Spark除了分区修剪之外，不使用分区信息进行查询优化。只使用扣饰。有关详细信息，请参见Spark是否知道DataFrame的分区键？。

结论：要有任何机会优化，你必须使用转移瘤和扣环。

一般来说，Spark不能优化“强类型”数据集上的操作。有关详细信息，请参见Spark2.0Dataset vs DataFrame，以及为什么谓词下推不用于类型化Dataset API（vs非类型化DataFrame API）？

正确的做法是：

>

用扣。

val n: Int
someDF.write.bucketBy(n, "date").saveAsTable("df")

删除函数式API，转而使用SQL API：

import org.apache.spark.sql.functions.explode

val df = spark.table("df")

val adf = df.select($"date", explode($"aRecords").alias("aRecords"))
val bdf = df.select($"date", explode($"bRecords").alias("bRecords"))

adf.join(bdf, Seq("date"))