Spark 2 选项数据集答案

【问题标题】：Spark 2 Datasets of OptionsSpark 2 选项数据集
【发布时间】：2016-11-09 15:11:07
【问题描述】：

我有一个字符串数据集，我使用一个可能失败的函数将其解析为案例类的数据集（例如，如果我尝试解析的数据不可用）。出于这个原因，该函数返回一个选项（Scala）。所以我最终得到了一个 Option[MyCaseClass] 的数据集。

Spark 似乎接受了该 Dataset 并对其进行处理，但如果解析失败，它不会返回 None 而是返回给我 Some(MyCaseClass(null, null...)) 。

这是一个这样做的代码示例：

recordsDs
  .map { record =>
    val maybeArticle = unmarshallArticle(record)
    if (maybeArticle.isEmpty) {
      println(s"Could not parse record $record into an article.")
    }
    maybeArticle
  }
  .filter(_.isDefined)
  .map(_.get)
  .collect().toList // Always returns a List(Some(Article(null, null), Some(Article...

这是一个说明案例的笔记本https://databricks-prod-cloudfront.cloud.databricks.com/public/4027ec902e239c93eaaa8714f173bcfc/4480125715694487/1289561535151709/7956941984681624/latest.html

我的猜测是，在序列化然后反序列化 Option 值时，Spark 使用 Some() 构造函数，而不是检查 Option 是 Some 还是 None。

我显然可以在我的对象周围创建一个包装器，例如 MaybeArticle(article: Option[Article]) ，但我想知道 Spark 是否可以正确处理 Options 的数据集？

【问题讨论】：

标签： scala apache-spark spark-dataframe scala-option

【解决方案1】：

我认为解决方案是使用flatMap。这是一个非常愚蠢的例子：

scala> val ds = Seq(("a1"), ("a2"), ("a4"), ("b1"), ("b2")).toDS
ds: org.apache.spark.sql.Dataset[String] = [value: string]

scala> ds.show 
+-----+        
|value|        
+-----+        
|   a1|        
|   a2|        
|   a4|        
|   b1|        
|   b2|        
+-----+        

scala> val ds2 = ds.flatMap{x => if (x.contains("a")) Some(x) else None}
ds2: org.apache.spark.sql.Dataset[String] = [value: string]

scala> ds2.show
+-----+
|value|
+-----+
|   a1|
|   a2|
|   a4|
+-----+

之所以可行，是因为Some 和None 的作用类似于可以使用flatMap 解包的集合（其中None 元素被省略）。

【讨论】：

谢谢！请注意，还有一个flatMapGroups 用于与ds.groupByKey 合作。