Spark数据集减少空值？答案

【问题标题】：Spark dataset reduce with null values?Spark数据集减少空值？
【发布时间】：2017-08-25 20:00:24
【问题描述】：

我正在使用此代码创建数据框：

  val data = List(
    List(444.1235D),
    List(67.5335D),
    List(69.5335D),
    List(677.5335D),
    List(47.5335D),
    List(null)
  )

  val rdd = sparkContext.parallelize(data).map(Row.fromSeq(_))
  val schema = StructType(Array(
    StructField("value", DataTypes.DoubleType, true)
  ))

  val df = sqlContext.createDataFrame(rdd, schema)

然后我将我的 udf 应用到它：

val multip: Dataset[Double] = df.select(doubleUdf(df("value"))).as[Double]

然后我尝试在这个数据集上使用 reduce：

val multipl = multip.reduce(_ * _)

结果我得到了 0.0。我也尝试过滤掉空值

val multipl = multip.filter(_ != null).reduce(_ * _)

结果相同。如果我从数据中删除空值，一切都会正常工作。如何使用 null 值进行 reduce 工作？

我的 udf 是这样定义的：

val doubleUdf: UserDefinedFunction = udf((v: Any) => Try(v.toString.toDouble).toOption)

【问题讨论】：

doubleUdf 的定义是什么。
@JustinPihony 我已将其添加到问题中

标签： scala apache-spark

【解决方案1】：

我将以一个强有力的假设来回答，即您的 doubleUdf 函数将值转换为双精度值，而不是使用 Option 包装器来处理空值，而是将空值转换为 0.0。因此，如果您想保留删除空值的逻辑，请在其他任何内容之前进行过滤：

df.na.drop.select(doubleUdf(df("value"))).as[Double]

【讨论】：

其实我的 udf 会按照这里的建议返回 Option [link]stackoverflow.com/questions/32357164/… 将其添加到问题中
赞成，但 select(doubleUdf(df("value"))) 在这里完全是多余的。数据已经是DoubleType，如果不是，最好建议cast，你不觉得吗？
这只是一个例子。此函数假设适用于其他数据类型

【解决方案2】：

首先，我想问你为什么还要和null打交道。我会评估我读取数据的方式，以确保不会发生这种情况。

然后我会注意到，您可以在达到 RDD 级别之前从内存中的 List 中消除 null，例如：

data.flatMap(Option(_)).flatten

但如果你必须在 RDD 级别处理null，你有选择（没有双关语）：

sparkContext.parallelize(data).filter(!_.contains(null))

或

sparkContext.parallelize(data).map(_.flatMap(Option(_))).filter(_.nonEmpty)

我更喜欢后者。我不喜欢在 Scala 代码中查看 null。

我会远离基于 UDF 的解决方案，因为 Spark 无法优化 UDF，而且因为像 null 这样蹩脚的东西而失去 Spark 的优化能力是一种耻辱。

【讨论】：