Google Cloud Dataflow 中的 FileBasedSource 使用示例答案

【问题标题】：Example of FileBasedSource usage in Google Cloud DataflowGoogle Cloud Dataflow 中的 FileBasedSource 使用示例
【发布时间】：2016-04-11 08:43:30
【问题描述】：

有人可以发布一个继承 FileBasedSource 的简单示例吗？我是 Google Dataflow 的新手，对 Java 非常缺乏经验。我的目标是在将行号作为键的同时读取文件，或者根据行号跳过行。

【问题讨论】：

标签： google-cloud-dataflow

【解决方案1】：

XMLSource 的实现是了解 FileBasedSource 工作原理的良好起点。您可能会希望您的读者可以使用类似的东西（其中 readNextLine() 读取到行尾并更新偏移量）：

protected void startReading(ReadableByteChannel channel) throws IOException {
  if (getCurrentSource().getMode() == FileBasedSource.Mode.SINGLE_FILE_OR_SUBRANGE) {
    // If we are not at the beginning of a line, we should ignore the current line.
    if (getCurrentSource().getStartOffset() > 0) {
      SeekableByteChannel seekChannel = (SeekableByteChannel) channel;
      // Start from one character back and read till we find a new line.
      seekChannel.position(seekChannel.position() - 1);
      nextOffset = seekChannel.position() + readNextLine(new ByteArrayOutputStream());
    }
  }
}

我用完整的LineIO 示例创建了一个要点，它可能比 XMLSource 更简单。

【讨论】：

老实说，我发现 XMLSource 示例很难理解，因为它将 XML 的复杂性添加到读取文件的问题上。我必须弄清楚如何“减少它”。 The answer to this question mentions a "LineIO example from documentation" 但我找不到。
LineIO 示例目前用于测试，仅在私有仓库中可见，但我可以探索发布它。
我发布了一个要点并修改了答案。