基于第一个数据帧 Java 中的列创建具有新列值的新 Spark DataFrame答案

【问题标题】：Creating a new Spark DataFrame with new column value based on column in first dataframe Java基于第一个数据帧 Java 中的列创建具有新列值的新 Spark DataFrame
【发布时间】：2016-09-02 14:18:57
【问题描述】：

这应该很简单，但是....使用 Spark 1.6.1.... 我有 A、B、C 列的 DataFrame #1。带有值：

然后我创建一个带有新列 D 的新数据框：

DataFrame df2 = df1.withColumn("D", df1.col("C"));

到目前为止一切顺利，但我实际上希望 D 列中的值是有条件的，即：

// pseudo code
if (col C = "A") the col D = "X"
else if (col C = "B") the col D = "Y"
else col D = "Z"

然后我将删除列 C 并将 D 重命名为 C。我尝试查看列函数，但似乎没有任何内容符合要求；我想过使用 df1.rdd().map() 并遍历行，但除了实际上没有设法让它工作之外，我有点认为 DataFrames 的全部意义在于远离 RDD 抽象？

不幸的是，我必须在 Java 中执行此操作（当然，使用 Java 的 Spark 不是最佳的！！）。似乎我错过了显而易见的事情，并且很高兴在提出解决方案时被证明是个白痴！

【问题讨论】：

标签： java apache-spark dataframe apache-spark-sql

【解决方案1】：

我相信您可以使用when 来实现这一目标。此外，您可能可以直接替换旧列。对于您的示例，代码将类似于：

import static org.apache.spark.sql.functions.*;

Column newCol = when(col("C").equalTo("A"), "X")
    .when(col("C").equalTo("B"), "Y")
    .otherwise("Z");

DataFrame df2 = df1.withColumn("C", newCol);

有关when的更多详细信息，请查看Column Javadoc。

【讨论】：

谢谢你 - 我确实盯着脸上的明显：s - 我缺少的是 sql 函数的静态导入，即：import static org.apache.spark.sql.functions。 *
@user1128482 对不起，我忘记导入了。很高兴知道你最终发现了。

【解决方案2】：

感谢 Daniel，我已经解决了这个问题 :)

缺少的部分是 sql 函数的静态导入

import static org.apache.spark.sql.functions.*;

我一定已经尝试过一百万种不同的使用 when 的方法，但是因为我没有进行导入而导致编译失败/运行时错误。一旦导入，丹尼尔的答案就在现场！

【讨论】：

【解决方案3】：

您也可以使用 udf 来完成相同的工作。只写一个简单的 if then else 结构体

import org.apache.spark.sql.functions.udf
val customFunct = udf { d =>
      //if then else construct
    }

val new_DF= df.withColumn(column_name, customFunct(df("data_column")))

【讨论】：