1

我正在尝试编写一个使用 SparkRunner 运行、从本地文件读取并写入 HDFS 的 Beam 管道。

这是一个最小的例子：

选项类 -

package com.mycompany.beam.hdfsIOIssue;

import org.apache.beam.runners.spark.SparkPipelineOptions;
import org.apache.beam.sdk.io.hdfs.HadoopFileSystemOptions;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.Validation;

public interface WritingToHDFSOptions extends PipelineOptions, SparkPipelineOptions, HadoopFileSystemOptions {

  @Validation.Required
  @Description("Path of the local file to read from")
  String getInputFile();
  void setInputFile(String value);

  @Validation.Required
  @Description("Path of the HDFS to write to")
  String getOutputFile();
  void setOutputFile(String value);

}

梁主类-

package com.mycompany.beam.hdfsIOIssue;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.FileBasedSink;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.PCollection;

public class WritingToHDFS {

  public static void main(String[] args) {
    PipelineOptionsFactory.register(WritingToHDFSOptions.class);

    WritingToHDFSOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
      .as(WritingToHDFSOptions.class);

    Pipeline p = Pipeline.create(options);

    buildPipeline(p, options);

    p.run();
  }

  static void buildPipeline(Pipeline p, WritingToHDFSOptions options) {
    PCollection<String> input = p.apply("ReadLines", TextIO.read().from(options.getInputFile()));

    ResourceId resource = FileBasedSink.convertToFileResourceIfPossible(options.getOutputFile());
    TextIO.Write write = TextIO.write().to(resource);
    input.apply("WriteLines", write);
  }
}

像这样运行它：

spark-submit test --master yarn --deploy-mode cluster --class com.mycompany.beam.hdfsIOIssue.WritingToHDFS my-project-bundled-0.1-SNAPSHOT.jar --runner=SparkRunner --inputFile=testInput --outputFile=hdfs://testOutput

我期望发生的事情：它读取本地 testInput 文件中的行并将它们写入我的 hdfs 主目录中的新文件名 testOutput。

实际发生的情况：据我所知，什么也没有。Spark 说作业成功完成，我在日志中看到了 Beam 步骤，但是没有一个名为 testOutput 的文件或目录写入 hdfs 或我的本地目录。也许它是在火花执行器节点上本地编写的，但我无权访问这些节点来检查。

我猜要么我使用错误的 TextIO 接口，要么我需要做更多的事情来配置文件系统，而不仅仅是将它添加到我的 PipelineOptions 接口。但我找不到解释如何做到这一点的文档。

4

1 回答 1

2

我认为您的选项应如下所示：

--inputFile=hdfs:///testInput --outputFile=hdfs:///testOutput

您可能还想等到管道完成后才能看到结果：

p.run().waitUntilFinish();

您可以在此处找到 HDFS 写入（Avro 文件）的简单完整工作示例

请注意（BEAM-2277），这也可能适用于您运行的 Beam 版本（它会引发错误）。您可以使用以下方法解决此问题：

TextIO.Write write = TextIO.write().to(resource)
  .withTempDirectory(FileSystems.matchNewResource("hdfs:///tmp/beam-test", true));

如果您在公共 GitHub 存储库上打包您的项目，我将对其进行测试并帮助您开始运行。

于 2018-11-05T19:01:20.617 回答