Spark map 和 flatMap 结果类型答案

【问题标题】：Spark map and flatMap result typesSpark map 和 flatMap 结果类型
【发布时间】：2015-06-04 10:53:23
【问题描述】：

看起来 map 和 flatMap 返回不同的类型。

mySchamaRdd.map( p => Row.fromSeq(...)) 返回 org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]，这是 applySchema 函数（或 spark 1.3 中的 createDataFrame）所必需的。

但是，mySchamaRdd.flatMap( p => Row.fromSeq(...) 返回 org.apache.spark.rdd.RDD[Any]，我无法调用 applySchema()。

如何在 flatMap() 之后使用 applySchema()？

一个例子（输入模式：名称，描述）

Bob, "Software developer"
John, "I like spaghetti"

结果：

Bob, Software
Bob, Developer
John, I
John, like
John, spaghetti

【问题讨论】：

如果你的 map 函数返回一个标量，为什么你必须使用 flatMap？如果您可能从 map 函数中为每个传入记录输出多个记录，或者偶尔不输出记录，则 flatMap 很有用。
对于每一行我想返回多行。它就像文本标记器。表 Name、AboutMe Bob、“Software developer”John、“I like spaghetti”的示例结果：Bob、Software Bob、Developer John、I John、like John、spaghetti
@Dmitry 如果将flatMap 的映射嵌入到Seq 中会怎样？像这样mySchamaRdd.flatMap( p => Seq(Row.fromSeq(...))) ？
@kaktusito 那么结果将只包含 2 项：1) Bob, Seq(...); 2）约翰，序列（...）。我希望看到所有 5 个项目。

标签： scala apache-spark

【解决方案1】：

也许我误解了您创建SchemaRDD 的方式，或者您误解了flatMap 的工作方式。你试过了吗？

mySchemaRDD.flatMap( p => p.getString(1).split(" +").map( x => Row((p.getString(0), x))))

我认为mySchamaRdd.flatMap( p => Row.fromSeq(...)) 不适合使用flatMap，因为您应该返回一系列嵌入的东西，以便扁平它并提取您真正想要返回的内容。在您的情况下，您将其嵌入在Row 中，而您要返回的结果类型是Row 本身。

【讨论】：

我想在一个相当大的模式中再添加一列。我使用map( word => p :+ word)（在最后一张地图中）。看起来此代码返回 Seq ，这是问题的根本原因。添加行后：map( word => Row(p :+ word)) 工作正常。感谢您的帮助！
PS：另一种解决方案 - 在 flatMap 之后再使用一个 map() 与 Row：mySchamaRdd.flatMap( p => Row.fromSeq(...).map(Row(_))