使用查找表中的 withColumn 动态添加新列

Question

我在 Java 8 中使用 spark-sql-2.4.1v。我有一个场景，我需要从查找表中动态添加一列。

我有带有列的数据框 A, B, C, ..., X,Y, Z

当少数（原始）列（例如：A、B、C）值为 null 时，我需要取/替换列（例如：X、Y、Z）值，否则取原始列值。 我将获取此映射信息作为业务逻辑的一部分。 如果是这种情况，我将遵循下面的硬编码代码

 Dataset<Row>  substitutedDs = ds
                  .withColumn("A",
                             when(col("A").isNull() , col("X").cast(DataTypes.StringType))
                             .otherwise(col("A").cast(DataTypes.StringType))
                          )
                  .withColumn("C",
                             when(col("C").isNull() , col("Z").cast(DataTypes.StringType))
                             .otherwise(col("C").cast(DataTypes.StringType))
                         

哪个工作正常。但我需要动态/可配置地执行此操作以避免硬编码。

我将得到包含“code”和“code_substitutes”列信息的查找表，如下所示

-------------------------
| Code | Code_Substitute |
-------------------------
  A         X
  B         Y
  C         Z
-------------------------

我需要在上面动态构造“submittedDs”，怎么做？

Answer 1

使用 Java8，您可以使用此 Stream.reduce() 重载：

final Dataset<Row> dataframe = ...;
final Map<String, String> substitutes = ...;

final Dataset<Row> afterSubstitutions = codeSubstitutes.entrySet().stream()
    .reduce(dataframe, (df, entry) ->
            df.withColumn(entry.getKey(), when(/* replace with col(entry.getValue()) when null */)),
            (left, right) -> { throw new IllegalStateException("Can't merge two dataframes. This stream should not be a parallel one!"); }
    );

组合器（最后一个参数）应该合并两个并行处理的数据帧（如果流是 parallel() 流），但我们根本不允许这样做，因为我们只是在 @ 上调用此逻辑987654324@流。

---

一个更具可读性/可维护性的版本涉及将上述逻辑提取到专用方法中的额外步骤，例如：

    // ...
    Dataset<Row> nullSafeDf = codeSubstitutes.entrySet().stream()
        .reduce(dataframe, this::replaceIfNull, this::throwingCombiner);
    // ...
}


private Dataset<Row> replaceIfNull(Dataset<Row> df, Map.Entry<String, String> substitution) {
    final String original = substitution.getKey();
    final String replacement = substitution.getValue();
    return df.withColumn(original, when(col(original).isNull(), col(replacement))
            .otherwise(col(original)));
}

private <X> X throwingCombiner(X left, X right) {
    throw new IllegalStateException("Combining not allowed");
}

Answer 2

在 Scala 中，我会这样做

val substitueMapping: Map[String, String] = ??? //this is your substitute map, this is small as it contains columns and their null substitutes

val df = ??? //this is your main dataframe 

val substitutedDf = substituteMapping.keys().foldLeft(df)((df, k) => {
    df.withColumn(k, when(col(k).isNull, col(substituteMapping(k))).otherwise(col(k)))
    //do approproate casting in above which you have done in post
})

我认为 foldLeft 在 Java 8 中不存在，您可以通过重复修改变量并在 substituteMapping 上进行迭代来模拟相同的情况。