通过从 Scala 中的另一个数据框中获取值，根据行类别在 spark 数据框中输入空值答案

【问题标题】：Imputing null values in spark dataframe, based on the row category, by fetching the values from another dataframe in Scala通过从 Scala 中的另一个数据框中获取值，根据行类别在 spark 数据框中输入空值
【发布时间】：2020-03-20 23:36:24
【问题描述】：

所以我有一个如下所示的数据框，它已存储为名为 mean_value_gn5 的临时视图，以便我可以在需要获取数据时使用 sql() 进行查询。

+-------+----+
|Species|Avgs|
+-------+----+
|    NO2|  43|
|    NOX|  90|
|     NO|  31|
+-------+----+

此数据框存储使用ceil() 函数四舍五入到最接近整数的“物种”的分类平均值。我需要使用这些分类平均值来估算我感兴趣的数据框clean_gn5 中列Value 的缺失值。我创建了一个新列 Value_imp，它将保存我的最后一列与估算值。

我尝试这样做：

clean_gn5 = clean_gn5.withColumn("Value_imp",
      when($"Value".isNull, sql("Select Avgs from mean_value_gn5 where Species = "+$"Species").select("Avgs").head().getLong(0).toInt)
        .otherwise($"Value"))

上述代码运行，但值被错误地估算，即对于包含 Species 为 NO 的行，估算值是 43 而不是 31。

在执行此操作之前，我首先检查是否能够通过执行以下操作正确获取值：

println(sql("Select Avgs from mean_value_gn5 where Species = 'NO'").select("Avgs").head().getLong(0))

在对Species 进行硬编码后，我能够正确获取该值，并且根据我的理解，$"Species" 应该可以帮助我获取与该特定行的Species 列对应的值。

我还认为我可能错过了硬编码Species 值周围的单引号，即'NO'。所以我尝试执行以下操作

 clean_gn5 = clean_gn5.withColumn("Value_imp",
      when($"Value".isNull, sql("Select Avgs from mean_value_gn5 where Species = '"+$"Species"+"'").select("Avgs").head().getLong(0).toInt)
        .otherwise($"Value"))

但这导致了以下异常。

Exception in thread "main" java.util.NoSuchElementException: next on empty iterator

我对 Spark 和 Scala 还很陌生。

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

假设clean_gn5 包含数据

+-------+-----+
|Species|Value|
+-------+-----+
|    NO2|  2.3|
|    NOX|  1.1|
|     NO| null|
|    ABC|  4.0|
|    DEF| null|
|    NOX| null|
+-------+-----+

使用左连接将clean_gn5 与mean_value_gn5 连接会导致

+-------+-----+----+
|Species|Value|Avgs|
+-------+-----+----+
|    NO2|  2.3|  43|
|    NOX|  1.1|  90|
|     NO| null|  31|
|    ABC|  4.0|null|
|    DEF| null|null|
|    NOX| null|  90|
+-------+-----+----+

在此数据框上，您可以对每行应用您在问题中已经给出的逻辑，结果是（在删除 Avgs 列之后）：

+-------+-----+---------+
|Species|Value|Value_imp|
+-------+-----+---------+
|    NO2|  2.3|      2.3|
|    NOX|  1.1|      1.1|
|     NO| null|     31.0|
|    ABC|  4.0|      4.0|
|    DEF| null|     null|
|    NOX| null|     90.0|
+-------+-----+---------+

代码：

clean_gn5.join(mean_value_gn5, Seq("Species"), "left")
  .withColumn("Value_imp", when('value.isNull, 'Avgs).otherwise('value))
  .drop("Avgs")
  .show()

【讨论】：