使用 Spark 和 java 编写 CSV 文件 - 处理空值和引号

Question

初始数据在 Dataset<Row> 中，我正在尝试写入管道分隔文件，我希望每个非空单元格和非空值都放在引号中。空值或空值不应包含引号

result.coalesce(1).write()
            .option("delimiter", "|")
            .option("header", "true")
            .option("nullValue", "")
            .option("quoteAll", "false")
            .csv(Location);

预期输出：

"London"||"UK"
"Delhi"|"India"
"Moscow"|"Russia"

电流输出：

London||UK
Delhi|India
Moscow|Russia

如果我将“quoteAll”更改为“true”，我得到的输出是：

"London"|""|"UK"
"Delhi"|"India"
"Moscow"|"Russia"

Spark 版本是 2.3，java 版本是 java 8

Answer 1

爪哇答案。CSV 转义不仅仅是在周围添加“符号。您应该处理”内部字符串。因此，让我们使用 StringEscapeUtils 并定义将调用它的 UDF。然后只需将 UDF 应用于每一列。

import org.apache.commons.text.StringEscapeUtils;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.spark.sql.functions.*;
import org.apache.spark.sql.expressions.UserDefinedFunction;
import org.apache.spark.sql.types.DataTypes;

import java.util.Arrays;

public class Test {

    void test(Dataset<Row> result, String Location) {
        // define UDF
        UserDefinedFunction escape = udf(
            (String str) -> str.isEmpty()?"":StringEscapeUtils.escapeCsv(str), DataTypes.StringType
        );
        // call udf for each column
        Column columns[] = Arrays.stream(result.schema().fieldNames())
                .map(f -> escape.apply(col(f)).as(f))
                .toArray(Column[]::new);

         // save the result
        result.select(columns)
                .coalesce(1).write()
                .option("delimiter", "|")
                .option("header", "true")
                .option("nullValue", "")
                .option("quoteAll", "false")
                .csv(Location);
    }
}

旁注：coalesce(1) 是一个糟糕的调用。它收集一个执行者的所有数据。您可以在生产中为巨大的数据集获取 executor OOM。