使用 RDD 进行理解时发出警告

Question

我在 for comprension 中使用 RDD 时收到警告，我不确定这是否是我做错了什么。如果我这样做：

val sc = new SparkContext(...)

val anRDD = sc.parallelize(List(
  ("a", List(1, 2, 3)), 
  ("b", List(4), 
  ("c", List(5, 6))
)

for {
  (someString, listOfInts) <- anRDD
  someInt <- listOfInts
} yield (someString, someInt)

然后我得到这个输出：

 warning: `withFilter' method does not yet exist on org.apache.spark.rdd.RDD[(String, List[Int])], using `filter' method instead
  (s, li) <- rl

但它仍然成功 return a FlatMappedRDD[(String, Int)]。难道我做错了什么？或者忽略此警告是否安全？

更新： 我也接受 for-comprehension 如何将这些操作转换为 map/flatMap/filter 调用的答案，因为我不认为会有需要任何过滤器或 withFilter 调用。我假设它等同于类似这样的东西：

anRDD.flatMap(tuple => tuple.map(someInt => (tuple._1, someInt)))

但这不包括任何过滤器或 withFilter 调用，这似乎是警告的来源。

哦，我正在使用 Spark 1.2.0，Scala 2.10.4，这些都在 REPL 中。

Answer 1

首先，我不是专家，但做了一些挖掘，这是我发现的：

我使用 -print 编译了代码（因为 JavaDecompiler 由于某种原因失败了），这将打印出删除了所有 Scala 特定功能的程序。在那里，我看到了：

test.this.anRDD().filter({
    (new anonymous class anonfun(): Function1)
  }).flatMap({
    (new anonymous class anonfun(): Function1)
  }, ClassTag.apply(classOf[scala.Tuple2]));

您会注意到 filter...所以，我查看了 anonfun：

public final boolean apply(Tuple2<String, List<Object>> check$ifrefutable)
  {
    Tuple2 localTuple2 = check$ifrefutable;
    boolean bool;
    if (localTuple2 != null) {
      bool = true;
    } else {
      bool = false;
    }
    return bool;
  }

所以，如果你把所有这些放在一起，似乎 filter 正在理解中发生，因为它过滤掉了任何不是 Tuple2 的东西。

而且，首选是使用 withFilter 而不是 filter（不知道为什么 atm）。您可以通过反编译常规列表而不是 RDD

来看到这一点

object test {
  val regList = List(
  ("a", List(1, 2, 3)), 
  ("b", List(4)),
  ("c", List(5, 6))
  )

val foo = for {
  (someString, listOfInts) <- regList
  someInt <- listOfInts
} yield (someString, someInt)
}

反编译为：

test.this.regList().withFilter({
  (new anonymous class anonfun(): Function1)
}).flatMap({
  (new anonymous class anonfun(): Function1)
}, immutable.this.List.canBuildFrom()).$asInstanceOf[List]();

所以，它是一回事，除了它在可以使用的地方使用 withFilter

Answer 2

在将其发送到理解之前在 RDD 中调用 collect()。

val collectedList = anRDD.collect
for {
  (someString, listOfInts) <- collectedList
  someInt <- listOfInts
} yield (someString, someInt)

使用 RDD 进行理解时发出警告

Warning while using RDD in for comprehension

scala

apache-spark

for-comprehension

rdd