Apache Beam TextIO.使用行号读取 [英] 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.

推荐答案

您可以使用 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.使用行号读取的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！