spark中熵的高效计算

Question

给定一个 RDD（数据）和一个索引字段列表来计算熵。执行以下流程时，在 2MB（16k 行）的源上计算单个熵值大约需要 5 秒。

def entropy(data: RDD[Array[String]], colIdx: Array[Int], count: Long): Double = { 
  println(data.toDebugString)
    data.map(r => colIdx.map(idx => r(idx)).mkString(",") -> 1)
        .reduceByKey(_ + _)
        .map(v => {
        val p = v._2.toDouble / count
        -p * scala.math.log(p) / scala.math.log(2)
      })
        .reduce((v1, v2) => v1 + v2)
}

debugString的输出如下：

(entropy,MappedRDD[93] at map at Q.scala:31 (8 partitions)
  UnionRDD[72] at $plus$plus at S.scala:136 (8 partitions)
    MappedRDD[60] at map at S.scala:151 (4 partitions)
      FilteredRDD[59] at filter at S.scala:150 (4 partitions)
        MappedRDD[40] at map at S.scala:124 (4 partitions)
          MapPartitionsRDD[39] at mapPartitionsWithIndex at L.scala:356 (4 partitions)
            FilteredRDD[27] at filter at S.scala:104 (4 partitions)
              MappedRDD[8] at map at X.scala:21 (4 partitions)
                MappedRDD[6] at map at R.scala:39 (4 partitions)
                  FlatMappedRDD[5] at objectFile at F.scala:51 (4 partitions)
                    HadoopRDD[4] at objectFile at F.scala:51 (4 partitions)
    MappedRDD[68] at map at S.scala:151 (4 partitions)
      FilteredRDD[67] at filter at S.scala:150 (4 partitions)
        MappedRDD[52] at map at S.scala:124 (4 partitions)
          MapPartitionsRDD[51] at mapPartitionsWithIndex at L.scala:356 (4 partitions)
            FilteredRDD[28] at filter at S.scala:105 (4 partitions)
              MappedRDD[8] at map at X.scala:21 (4 partitions)
                MappedRDD[6] at map at R.scala:39 (4 partitions)
                  FlatMappedRDD[5] at objectFile at F.scala:51 (4 partitions)
                    HadoopRDD[4] at objectFile at F.scala:51 (4 partitions),colIdex,13,count,3922)

如果我再次收集 RDD 和 parallelize 需要大约 150 毫秒来计算（对于一个简单的 2MB 文件来说这似乎仍然很高） - 并且在处理时显然会产生挑战具有多个 GB 数据。正确使用 Spark 和 Scala 我缺少什么？

我最初的实现（表现更差）：

data.map(r => colIdx
  .map(idx => r(idx)).mkString(","))
  .groupBy(r => r)
  .map(g => g._2.size)
  .map(v => v.toDouble / count)
  .map(v => -v * scala.math.log(v) / scala.math.log(2))
  .reduce((v1, v2) => v1 + v2)

Answer 1

首先，您的代码中似乎存在错误，您需要处理0 中的p，因此-p * math.log(p) / math.log(2) 应该是if (p == 0.0) 0.0 else -p * math.log(p) / math.log(2)。

其次，您可以使用底数 e，您实际上不需要底数为 2。

无论如何，您的代码运行缓慢的原因可能是分区太少。每个 CPU 至少应该有 2-4 个分区，实际上我经常使用更多。你有多少个 CPU？

现在可能花费最长时间的不是熵计算，因为它非常简单 - 而是在 String 键上完成的 reduceByKey。是否可以使用其他数据类型？ colIdx 实际上是什么？ r 究竟是什么？

最后一个观察结果是你用这个colIdx.map(r.apply) 多次索引每条记录......你知道如果r 不是Array 或IndexedSeq 类型，这将非常慢......如果它是@ 987654331@ 这将是 O(index) 因为你必须遍历列表才能得到你想要的索引。