使用 Java 在 Spark 中将新列附加到现有 CSV 文件答案

【问题标题】：Appending a new column to existing CSV file in Spark with Java使用 Java 在 Spark 中将新列附加到现有 CSV 文件
【发布时间】：2016-02-23 07:58:55
【问题描述】：

我在这里找到了解决问题的方法Create new column with function in Spark Dataframe

但我很难将以下代码转换为 Java，因为它在 Scala 中

import org.apache.spark.sql.functions._
val myDF = sqlContext.parquetFile("hdfs:/to/my/file.parquet")
val coder: (Int => String) = (arg: Int) => {if (arg < 100) "little" else "big"}
val sqlfunc = udf(coder)
myDF.withColumn("Code", sqlfunc(col("Amt")))

谁能为此提供我的 Java 等效代码？我被困在 2 行以下的转换中

val coder: (Int => String) = (arg: Int) => {if (arg < 100) "little" else "big"}
val sqlfunc = udf(coder)

谢谢，

【问题讨论】：

好吧，显然你可以转换其中的一部分。因此，与其要求提供代码翻译服务，不如指出是哪些特定位给您造成了问题。
@TheArchetypalPaul，第 3 行和第 4 行让我很困扰，你能提供我的翻译吗？谢谢
您使用的是 Java 8 吗？

标签： java scala apache-spark

【解决方案1】：

创建您的用户定义函数：

public class CodeUdf implements  UDF1<Integer, String>{
    @Override
    public String call(Integer integer) throws Exception {
        if(integer < 100)
            return "little";
        else
            return"big";
    }
}

告诉 Spark

sqlContext.udf().register("Code", new CodeUdf(), DataTypes.IntegerType);

在选择中使用它。

df.selectExpr("value", "Code(value)").show();

【讨论】：

【解决方案2】：

import org.apache.spark.sql.functions._
val myDF = sqlContext.parquetFile("hdfs:/to/my/file.parquet")
//val coder: (Int => String) = (arg: Int) => {if (arg < 100) "little" else "big"}
//val sqlfunc = udf(coder)
myDF.selectExpr("Code", "case when Amt < 100 'little' else 'big' end ")

【讨论】：