在使用 RDD 的 Scala 中，如果 RDD[(k,Iterable[v]) 如何在 Iterable 中获取应用函数答案

【问题标题】：in Scala using RDD , how do you get apply function in the Iterable if RDD[(k,Iterable[v])在使用 RDD 的 Scala 中，如果 RDD[(k,Iterable[v]) 如何在 Iterable 中获取应用函数
【发布时间】：2016-02-25 03:46:03
【问题描述】：

我正在尝试寻找一种方法来有效地在 Iterable[Value] 上为给定的 RDD[Key,Iterable[Value]] 类应用过滤器（多个）。

原因是我想过滤 RDD 并最终找到与过滤器匹配的键

RDD 示例

 000473643-02,CompactBuffer((glucose,80.0), (glucose,80.0), (glucose2,80.0),   (fasting blood glucose,80.0), (glucose,80.0), (glucose,80.0), (glucose,80.0), (glucose,80.0)))
 (713003448-01,CompactBuffer((glucose,80.0), (glucose,80.0)))
 (000023838-01,CompactBuffer((glucose,80.0), (glucose,80.0)))
 (000772974-01,CompactBuffer((glucose,80.0), (glucose,80.0), (glucose,80.0)))
 (380670000-01,CompactBuffer((glucose,80.0), (glucose,80.0)))

所以在这种情况下，我只需要在满足以下条件时才输出 Key：

    glucose value is >= 80 or fasting blood glucose >= 80

【问题讨论】：

标签： scala rdd

【解决方案1】：

我会使用这样的东西：

case class ExceedsCondition(threshold:Double) {
  def violates(value:Double) = value >= threshold
}

val conditionsBroadcast = sc.broadcast(
  Map("glucose" -> ExceedsCondition(80.0), 
      "fasting-glucose" -> ExceedsCondition(81.0))
  )

val rdd = sc.parallelize(List("key1" -> List(("glucose" -> 83.0))))

val result = rdd.filter { case (_, xs) =>
            val conditions = conditionsBroadcast.value
            xs.exists { case (key, value) =>
                conditions.get(key).exists(_.violates(value))
            }
}

result.take(10)

【讨论】：

广播对所有条件进行 OR，有没有办法对所有条件进行 AND，其次我们可以做不同的条件 .. 像一个做 > 另一个 >= 在一个条件广播语句或我需要在同一个 RDD 上进行多次传递当我尝试不同的条件时，我收到错误：错误：值违反不是具有可序列化 xs.exists { case (key, value) => conditions.get(key) 的产品的成员.exists(_.violates(value)) } } @vitalii
哦，对了，我忘了问，上面的条件意味着条件都是 OR .. 有没有可能用不同的条件集（=。