google-cloud-dataflow - 带有行号的 Apache Beam TextIO.Read

Question

是否可以通过从 TextIO.Read 读入 PCollection 的行访问行号？对于这里的上下文，我正在处理一个 CSV 文件并且需要访问给定行的行号。

如果无法通过 TextIO.Read 使用某种自定义 Read 或转换似乎应该是可能的，但我无法弄清楚从哪里开始。

Answer 1

您可以使用 FileIO 手动读取文件，从 ReadableFile 读取时可以确定行号.

一个简单的解决方案如下所示:

p
    .apply(FileIO.match().filepattern("/file.csv"))
    .apply(FileIO.readMatches())
    .apply(FlatMapElements
            .into(strings())
            .via((FileIO.ReadableFile f) -> {
                List<String> result = new ArrayList<>();
                try (BufferedReader br = new BufferedReader(Channels.newReader(f.open(), "UTF-8"))) {
                    int lineNr = 1;
                    String line = br.readLine();
                    while (line != null) {
                        result.add(lineNr + "," + line);
                        line = br.readLine();
                        lineNr++;
                    }
                } catch (IOException e) {
                    throw new RuntimeException("Error while reading", e);
                }
                return result;
            }));

上面的解决方案只是将行号添加到每个输入行。