scala - 如何在Spark中按键分区RDD？

Question

鉴于 HashPartitioner 文档说:

[HashPartitioner] implements hash-based partitioning using Java's Object.hashCode.

说我要分区 DeviceData由其 kind .

case class DeviceData(kind: String, time: Long, data: String)

对 RDD[DeviceData] 进行分区是否正确？通过覆盖 deviceData.hashCode()方法并仅使用 kind 的哈希码?

但鉴于 HashPartitioner需要多个分区参数我很困惑我是否需要提前知道种类的数量以及如果种类多于分区会发生什么？

如果我将分区数据写入磁盘，它在读取时将保持分区状态是否正确？

我的目标是打电话

  deviceDataRdd.foreachPartition(d: Iterator[DeviceData] => ...)

并且只有DeviceData的相同 kind迭代器中的值。

Answer 1

做一个groupByKey怎么样？使用 kind .或者另一个 PairRDDFunctions方法。

您让我觉得您并不真正关心分区，只是在一个处理流程中获得所有特定类型？

配对函数允许:

rdd.keyBy(_.kind).partitionBy(new HashPartitioner(PARTITIONS))
   .foreachPartition(...)

但是，您可能会更安全一点:

rdd.keyBy(_.kind).reduceByKey(....)

或 mapValues或许多其他配对函数，可确保您将碎片作为一个整体