apache-spark - 分组数据的 Spark 并行处理

Question

最初，我有很多数据。但是使用 spark-SQL 尤其是 groupBy 可以将其缩减到可管理的大小。 (适合单个节点的RAM)

我该怎么办 对所有组执行功能(并行) (分布在我的节点之间)？

我如何确保 的数据单个组收集到单个节点 ?例如。我可能想使用 local matrix用于计算，但不想遇到有关数据局部性的错误。

Answer 1

假设你有 x 没有。执行程序(在您的情况下每个节点可能有 1 个执行程序)。并且您希望以这样的方式对键上的数据进行分区，使每个键都落入一个独特的存储桶中，这将是一个完美的分区器。没有通用的方法这样做，但如果有一些特定于您的数据的固有分布/逻辑，则有可能实现这一点。
我处理过一个特定的案例，我发现 Spark 的内置哈希分区器在均匀分配 key 方面做得不好。所以我使用 Guava 编写了一个自定义分区器，如下所示:

  class FooPartitioner(partitions: Int) extends org.apache.spark.HashPartitioner(partitions: Int) {
    override def getPartition(key: Any): Int = {
      val hasherer = Hashing.murmur3_32().newHasher()
      Hashing.consistentHash(
        key match {
          case i: Int => hasherer.putInt(i).hash.asInt()
          case _ => key.hashCode
          },PARTITION_SIZE)
  }
 }

然后我将此分区器实例添加为我正在使用的 combineBy 的参数，以便生成的 rdd 以这种方式进行分区。
这在将数据分发到 x 个桶方面做得很好，但我想不能保证每个桶只有 1 个键。

如果您使用的是 Spark 1.6 并使用数据帧，您可以像这样定义一个 udf val hasher = udf((i:Int)=>Hashing.consistentHash(Hashing.murmur3_32().newHasher().putInt(i) .hash.asInt(),PARTITION_SIZE))并做 dataframe.repartition(hasher(keyThatYouAreUsing)) 希望这提供了一些开始的提示。