我正在尝试计算 pointwise mutual information (PMI)。
我在这里分别为 p(x, y) 和 p(x) 定义了两个 RDD:
pii: RDD[((String, String), Double)]
pi: RDD[(String, Double)]
我正在编写的任何用于从 RDD 计算 PMI 的代码
pii
和 pi
不漂亮。我的做法是先把RDD拉平pii
并加入 pi
两次同时按摩元组元素。val pmi = pii.map(x => (x._1._1, (x._1._2, x._1, x._2)))
.join(pi).values
.map(x => (x._1._1, (x._1._2, x._1._3, x._2)))
.join(pi).values
.map(x => (x._1._1, computePMI(x._1._2, x._1._3, x._2)))
// pmi: org.apache.spark.rdd.RDD[((String, String), Double)]
...
def computePMI(pab: Double, pa: Double, pb: Double) = {
// handle boundary conditions, etc
log(pab) - log(pa) - log(pb)
}
显然,这很糟糕。有没有更好的(惯用的)方法来做到这一点?
注意:我可以通过将 log-probs 存储在
pi
中来优化日志。和 pii
但选择这样写是为了让问题保持清晰。
最佳答案
使用 broadcast
将是一个解决方案。
val bcPi = pi.context.broadcast(pi.collectAsMap())
val pmi = pii.map {
case ((x, y), pxy) =>
(x, y) -> computePMI(pxy, bcPi.value.get(x).get, bcPi.value.get(y).get)
}
假设:
pi
拥有所有 x
和 y
在 pii
.
关于apache-spark - 在 Spark 中计算逐点互信息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/29620297/