使用数据流将数据导出到 csv 时如何将列名称添加为 header

Question

我正在通过 Dataflow 将一些数据导出到 csv，但除了数据之外，我想将每个列名称添加为输出文件的第一行，例如

    col_name1, col_name2, col_name3, col_name4 ...
    data1.1, data1.2, data1.3, data1.4 ...
    data2.1 ...

与当前 API 有什么关系吗？（搜索了 TextIO.Write 但没有发现任何相关的东西...）或者无论如何我可以 "insert" to-be-exported PCollection 头部的列名并强制按顺序写入数据...？

Answer 1

TextIO.Write 没有内置方法可以做到这一点。 PCollection 是无序的，因此不可能在前面添加元素。您可以编写一个自定义 BoundedSink 来执行此操作。

Answer 2

Custom sink APIs 现在可用，如果您想成为制作 CSV 接收器的勇敢者。当前的解决方法将输出构建为单个字符串并在完成捆绑时将其全部输出：

PCollection<String> output = data.apply(ParDo.of(new DoFn<String, String>() {
private static final long serialVersionUID = 0;

String new_line = System.getProperty("line.separator");
String csv_header = "id, stuff1, stuff2, stuff3" + new_line;
StringBuilder csv_body = new StringBuilder().append(csv_header);

@Override
public void processElement(ProcessContext c) {
    csv_body.append(c.element()).append(newline);
}

@Override
public void finishBundle(Context c) throws Exception {
    c.output(csv_body);
}

})).apply(TextIO.Write.named("WriteData").to(options.getOutput()));

这仅在您的 BIG 输出字符串适合内存时才有效

Answer 3

从 Dataflow SDK 版本 1.7.0 开始，您有 withHeader function in TextIO.Write .

所以你可以这样做：

TextIO.Write.named("WriteToText")
            .to("/path/to/the/file")
            .withHeader("col_name1,col_name2,col_name3,col_name4")
            .withSuffix(".csv"));

新行字符自动添加到 header 的末尾。

使用数据流将数据导出到 csv 时如何将列名称添加为 header

How to add column name as header when using dataflow to export data to csv

google-cloud-dataflow