Apache Beam Getting Started - #3

readfilefn.java

public class ReadFileFn extends DoFn<FileIO.ReadableFile, KV<Metadata, String>> {

    @ProcessElement
    public void processElement(ProcessContext c) {
        try {
            FileIO.ReadableFile f = c.element();
            String filename = f.getMetadata().resourceId().getFilename();

            ReadableByteChannel rbc = f.open();
            try (InputStream stream = Channels.newInputStream(rbc);
                 BufferedReader br = new BufferedReader(new InputStreamReader(stream))) {

                long lineId = 1;
                String line;
                while ((line = br.readLine()) != null) {
                    c.output(KV.of(Metadata.of(filename, lineId), line));

                    lineId++;
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}