Spark --错误：类型不匹配；发现 : (Int, String) 需要： TraversableOnce[?]

Question

我是 spark 编程和 scala 的新手，我无法理解 map 和 flatMap 之间的区别。使用flatMap时，为什么方法中使用的“选项”工作正常

def parseNames(line: String) : Option[(Int,String)]  = {
  var fields = line.split('\"')
  if (fields.length >1) {
    return Some(fields(0).trim().toInt,fields(1) )
  }
  else {
    return None
  }
}
def main(args: Array[String]) {
  val sc = new SparkContext("local[*]","DemoHero")
  val txt= sc.textFile("../marvel-names1.txt")  
  val rdd = txt.flatMap(parseNames)

但没有“选项”，就会报错：

def parseNames(line: String) : (Int, String)  = {
  var fields = line.split('\"')    
  (fields(0).trim().toInt,fields(1) )
}

def main(args: Array[String]) {
  val sc = new SparkContext("local[*]","DemoHero")   
  val txt= sc.textFile("../marvel-names1.txt")  
  val rdd = txt.flatMap(parseNames)

据我了解，flatmap 将 Rdd 放入 String/Int Rdd 的集合中。我在想，在这种情况下，两者都应该没有任何错误。请让我知道我在哪里犯了错误。

Answer 1

TL;DR：有一个从 Option 到 Iterable 的隐式转换，这就是为什么你的第一个 flatMap 不会失败。

---

从inheritance hierarchy of Option 来看，完全不清楚为什么 RDD 的 flatMap 期望 返回类型中带有TraversableOnce 的参数将接受一个返回Option 的函数，因为 Option 不扩展 TraversableOnce。

但是，如果您打印由您的flatMap 生成的脱糖代码，则会出现以下合成函数定义：

@SerialVersionUID(value = 0) final <synthetic> class anonfun$1 extends scala.runtime.AbstractFunction1 with Serializable {
  final def apply(line: String): Iterable = scala.this.Option.option2Iterable(org.example.ClassName.parseNames$1(line));
  final <bridge> <artifact> def apply(v1: Object): Object = anonfun$1.this.apply(v1.$asInstanceOf[String]());
  def <init>(): <$anon: Function1> = {
    anonfun$1.super.<init>();
    ()
  }
}

细节并不那么重要，它需要一个line: String 并返回一个Iterable。 有趣的是Option.option2Iterable 部分。

这是一个隐式转换defined directly on Option， 它悄悄地将选项转换为Iterable，而Iterable 是TraversableOnce 的一个特例。

这就是编译器如何将option2Iterable 潜入合成Function-definition 在您的方法和flatMap 的调用之间进行调解。现在你有一个论点 键入String => Iterable[(Int, String)]，这样flatMap 就可以正常编译了。

请注意，如果没有包含您的方法的合成 Function-instance，它将无法工作。如果你这样声明parseNames：

def parseNames: String => Option[(Int,String)] = { line => 

这将是一个简单的编译器错误。

---

您的第二个代码 sn-p 不应该编译，幸运的是，它确实没有：pairs 不是 Traversable，所以 flatMap 不接受 parseNames(line: String) : (Int, String) 作为参数。你想在这里使用的是 map，因为你想映射每个字符串到一对(Int, String)。

flatMap 用于不同的用例：它用于将原始集合中的每个元素转换为 另一个集合，然后将所有生成的集合扁平化为一个集合，例如，

sc.parallelize(List(1, 2, 3)).flatMap{ x => List(x, x*x, x*x*x) }

首先会为每个x 生成一个TraversableOnce：

List(1,1,1)
List(2,4,8)
List(3,9,27)

然后把它们粘在一起，这样你就可以得到一个带有条目的RDD

1,1,1,2,4,8,3,9,27

它以相同的方式与Option 一起使用，因为“从道德上讲”它就像一个包含 0 到 1 元素的列表，即使它没有在其继承层次结构中明确说明这一点。

---

注意关于“不应该编译”的表述：每当我写（你的代码或其他代码）“不应该编译”时，我并不是说我一般希望你您的代码中有编译错误。我的意思是如果代码有问题，编译器应该尽快产生清晰的错误信息。

Answer 2

def parseNames (line: String): Option[(Int,String)]  = {
  var fields = line.split('\"')
  if (fields.length > 1) {
    Some (fields(0).trim ().toInt, fields(1))
  }
  else {
    None
  }
}

（去掉了嘈杂的“返回”）

那么，None 什么时候返回？如果 fields.length 不 > 1。如果没有至少 2 个字段（fields(0) 和 fields(1)），fields(0).trim().toInt 可能会成功，但fields(1) 会失败。

Answer 3

flatMap 需要 iterables 作为被调用函数的返回类型。因为flatMap 会遍历 iterable 的每个元素并返回每个扁平化的元素。

在您的第一个parseNames 函数中，返回Option[(Int, String)]，它是一个容器，由于使用了隐式函数，它可以像iterable 一样运行。所以flatMap 工作了。

但是在您的第二个parseNames 中，Tuple2[Int, String] 被返回，它不是可迭代的。因为Tuple2 不能被迭代，但是元素可以通过使用_1 和_2 得到。所以flatMap 显示了编译错误。

我希望解释清楚。

如果您将Array 作为返回，第二个parseNames 会起作用

def parseNames(line: String) : Array[(Int, String)]  = {
  var fields = line.split('\"')

  Array((fields(0).trim().toInt,fields(1)))
}

或List 为

def parseNames(line: String) : List[(Int, String)]  = {
  var fields = line.split('\"')

  List((fields(0).trim().toInt,fields(1)))
}

或Seq 为

def parseNames(line: String) : Seq[(Int, String)]  = {
  var fields = line.split('\"')

  Seq((fields(0).trim().toInt,fields(1)))
}

因为它们都是 iterables，就像 Option 一样。