使用 Google Cloud Dataflow 生成大文件

Question

我正在运行以下数据流代码作为 JUnit 测试类的一部分

@Test
public void dataFlowGenerator() {
  DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class);
  options.setRunner(BlockingDataflowPipelineRunner.class);
  options.setStagingLocation("gs://mybucket/lt");
  options.setProject("myProject");
  Pipeline p = Pipeline.create(options);

  List<String> list = Arrays.asList("sup1", "sup2", "sup3");
  p.apply(Create.of(list)).apply(ParDo.of(new generate())).apply(
      TextIO.Write.to("gs://mybucket/lt/df.txt"));
}


private class generate extends DoFn<String, String> implements Serializable {

  @Override
  public void processElement(ProcessContext c) throws Exception {
    new DoFn<String, String>() {
      @Override
      public void processElement(ProcessContext c) {
        c.output(c.element());
      }
    };
  }
}

这不会在云存储中提到的文件中输出任何内容（只是一个空文件 df.txt-00000-of-00001）。我期待 processElement 会为列表中的每个字符串项调用，它们将被输出到输出文件。

如何使用随机字符串生成数百万条记录。在这种情况下，数据流实际上不需要任何输入源。它应该能够只转换一些种子字符串和输出。

Answer 1

两件事：

首先，您根本不需要这样做ParDo。您的转换只是身份转换 - 您可以将 Create 的输出直接输入到 TextIO.Write。

但我还想澄清为什么您没有看到 goutput：您的 DoFn 子类有一个 processElement 方法，该方法创建另一个 DoFn 并丢弃该值。你会想写这个：

private class Generate extends DoFn<String, String> implements Serializable {

  @Override
  public void processElement(ProcessContext c) throws Exception {
    c.output(c.element());
  }
}