带有行号的 Apache Beam TextIO.Read [英] Apache Beam TextIO.Read with line number

问题描述

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

Is it possible to get access to line numbers with the lines read into the PCollection from TextIO.Read? For context here, I'm processing a CSV file and need access to the line number for a given line.

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

If not possible through TextIO.Read it seems like it should be possible using some kind of custom Read or transform, but I'm having trouble figuring out where to begin.

推荐答案

您可以使用 FileIO 手动读取文件，从ReadableFile.

You can use FileIO to read the file manually, where you can determine the line number when you read from the ReadableFile.

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

A simple solution can look as follows:

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;
            }));

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

The solution above just prepends the line number to each input line.

这篇关于带有行号的 Apache Beam TextIO.Read的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！