spark scala 条件连接替换空值

Question

我有两个数据框。我想使用来自col1 的df2 的值替换col1 的df1 中的值为空的值。请记住，df1 与df2 类似，可以有 > 10^6 行，并且df1 有一些额外的列，这些列与df2 的一些额外列不同。

我知道如何加入，但我不知道如何在 Spark 中使用 Scala 进行某种条件加入。

df1

name   | col1 | col2 | col3
----------------------------
foo    | 0.1  | ...
bar    | null |
hello  | 0.6  |
foobar | null |


df2 

name   | col1 | col7
--------------------
lorem  | 0.1  |
bar    | 0.52 |
foobar | 0.47 |

编辑：

这是我目前的解决方案：

df1.select("name", "col2", "col3").join(df2, (df1("name") === df2("name")), "left").select(df1("name"), col("col1"))

EDIT2：

val df1 = Seq(
  ("foo", Seq(0.1), 10, "a"),
  ("bar", Seq(), 20, "b"),
  ("hello", Seq(0.1), 30, "c"),
  ("foobar", Seq(), 40, "d")
).toDF("name", "col1", "col2", "col3")

val df2 = Seq(
  ("lorem", Seq(0.1), "x"),
  ("bar", Seq(0.52), "y"),
  ("foobar", Seq(0.47), "z")
).toDF("name", "col1", "col7")

display(df1.
  join(df2, Seq("name"), "left_outer").
  select(df1("name"), coalesce(df1("col1"), df2("col1")).as("col1")))

返回：

name   | col1
bar    | []
foo    | [0.1]
foobar | []
hello  | [0.1]

Answer 1

在执行left join 之后，考虑在col1 上使用coalesce。要按照 cmets 部分的修订要求同时处理 nulls 和空 arrays（在 ArrayType 的情况下），使用 when/otherwise 子句，如下所示：

val df1 = Seq(
  ("foo",    Some(Seq(0.1)), 10, "a"),
  ("bar",    None,           20, "b"),
  ("hello",  Some(Seq(0.1)), 30, "c"),
  ("foobar", Some(Seq()),    40, "d")
).toDF("name", "col1", "col2", "col3")

val df2 = Seq(
  ("lorem",  Seq(0.1),  "x"),
  ("bar",    Seq(0.52), "y"),
  ("foobar", Seq(0.47), "z")
).toDF("name", "col1", "col7")

df1.
  join(df2, Seq("name"), "left_outer").
  select(
    df1("name"),
    coalesce(
      when(lit(df1.schema("col1").dataType.typeName) === "array" && size(df1("col1")) === 0, df2("col1")).otherwise(df1("col1")), 
      df2("col1")
    ).as("col1")
  ).
  show
/*
+------+------+
|  name|  col1|
+------+------+
|   foo| [0.1]|
|   bar|[0.52]|
| hello| [0.1]|
|foobar|[0.47]|
+------+------+
*/

---

更新：

令人惊讶的是，Spark 似乎不像大多数其他语言那样处理conditionA && conditionB——即使conditionA 为假conditionB 仍将被评估，并且仍将&& 替换为嵌套的when/otherwise不会解决问题。这可能是由于内部翻译的case/when/else SQL 的执行方式受到限制。

因此，当 col1 是非 ArrayType 时，通过特定于数组的函数 size() 进行的上述 when/otherwise 数据类型检查失败。鉴于此，我会放弃动态列类型检查并根据 col1 是否为 ArrayType 执行不同的查询，假设它是预先知道的：

df1.
  join(df2, Seq("name"), "left_outer").
  select(
    df1("name"),
    coalesce(
      when(size(df1("col1")) === 0, df2("col1")).otherwise(df1("col1")),  // <-- if col1 is an array
      // df1("col1"),  // <-- if col1 is not an array
      df2("col1")
    ).as("col1")
  ).
  show