Cloud Dataflow 中的 ETL 和解析 CSV 文件答案

【问题标题】：ETL & Parsing CSV files in Cloud DataflowCloud Dataflow 中的 ETL 和解析 CSV 文件
【发布时间】：2016-05-25 14:33:33
【问题描述】：

我是云数据流和 Java 的新手，所以我希望这是一个正确的问题。

我有一个 csv 文件，其中包含 n 个列和行，可以是字符串、整数或时间戳。我需要为每列创建一个新的 PCollection 吗？

我在示例中找到的大多数文档都类似于以下内容：

PCollection<String> data = p.apply(TextIO.Read.from("gs://abc/def.csv"));

但对我来说，将整个 csv 文件作为字符串导入是没有意义的。我在这里缺少什么，我应该如何设置我的 PCollections？

【问题讨论】：

标签： csv google-cloud-dataflow

【解决方案1】：

line.split(",");

如果这样的行数据，String.split 没有意义：

a,b,c,"we,have a string contains comma",d,e

处理csv数据的一种属性方式是导入一个csv库：

        <dependency>
            <groupId>com.opencsv</groupId>
            <artifactId>opencsv</artifactId>
            <version>3.7</version>
        </dependency>

并在 ParDo 中使用以下代码：

public void processElement(ProcessContext c) throws IOException {
    String line = c.element();
    CSVParser csvParser = new CSVParser();
    String[] parts = csvParser.parseLine(line);
}

【讨论】：

【解决方案2】：

此示例将创建一个集合，该集合包含文件中每行 1 个 String，例如如果文件是：

Alex,28,111-222-3344
Sam,30,555-666-7788
Drew,19,123-45-6789

那么集合逻辑上将包含"Alex,28,111-222-3344"、"Sam,30,555-666-7788" 和"Drew,19,123-45-6789"。您可以通过 ParDo 或 MapElements 转换对集合进行管道化，从而在 Java 中应用进一步的解析代码，例如：

class User {
    public String name;
    public int age;
    public String phone;
}

PCollection<String> lines = p.apply(TextIO.Read.from("gs://abc/def.csv"));
PCollection<User> users = lines.apply(MapElements.via((String line) -> {
    User user = new User();
    String[] parts = line.split(",");
    user.name = parts[0];
    user.age = Integer.parseInt(parts[1]);
    user.phone = parts[2];
    return user;
}).withOutputType(new TypeDescriptor<User>() {});)

【讨论】：

谢谢@jkff，我现在看看。
我会使用 String[] parts = line.split(",(?=([^\"]*\"[^\"]*\")*[^\" ]*$)") 来解释额外的逗号和引号。
这太好了，谢谢。是否有更强大的方法来处理具有意外列顺序的 CSV？
@Eugene 在您的示例中是否可以使用 lines.apply(ParDo.of(new CreateUser()) 并在 DoFn processElement 方法中编写转换逻辑？以及 withOutputType() 的用途是什么。
ParDo 也可以，只是不够简洁。由于与 Java 类型擦除相关的技术原因，需要 withOutputType，如果您省略它，编码器推断在这种情况下将不起作用。