Apache Beam 是否支持其输出的自定义文件名？

Question

虽然在分布式处理环境中通常使用“part-000”等“part”文件名，但是否可以编写某种扩展名来重命名单个输出文件名 （例如每个窗口的文件名）的 Apache Beam？

为此，可能必须能够为窗口指定名称或根据窗口的内容推断文件名。我想知道这种方法是否可行。

至于解决方案应该是流式还是批处理，最好是流式模式示例

Answer 1

是的，正如 jkff 所建议的，您可以使用 TextIO.write.to(FilenamePolicy) 来实现这一点。

示例如下：

如果您想将输出写入特定的本地文件，您可以使用：

lines.apply(TextIO.write().to("/path/to/file.txt"));

下面是使用前缀link 编写输出的简单方法。此示例适用于 google 存储，您可以使用本地/s3 路径代替此示例。

public class MinimalWordCountJava8 {

  public static void main(String[] args) {
    PipelineOptions options = PipelineOptionsFactory.create();
    // In order to run your pipeline, you need to make following runner specific changes:
    //
    // CHANGE 1/3: Select a Beam runner, such as BlockingDataflowRunner
    // or FlinkRunner.
    // CHANGE 2/3: Specify runner-required options.
    // For BlockingDataflowRunner, set project and temp location as follows:
    //   DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class);
    //   dataflowOptions.setRunner(BlockingDataflowRunner.class);
    //   dataflowOptions.setProject("SET_YOUR_PROJECT_ID_HERE");
    //   dataflowOptions.setTempLocation("gs://SET_YOUR_BUCKET_NAME_HERE/AND_TEMP_DIRECTORY");
    // For FlinkRunner, set the runner as follows. See {@code FlinkPipelineOptions}
    // for more details.
    //   options.as(FlinkPipelineOptions.class)
    //      .setRunner(FlinkRunner.class);

    Pipeline p = Pipeline.create(options);

    p.apply(TextIO.read().from("gs://apache-beam-samples/shakespeare/*"))
     .apply(FlatMapElements
         .into(TypeDescriptors.strings())
         .via((String word) -> Arrays.asList(word.split("[^\\p{L}]+"))))
     .apply(Filter.by((String word) -> !word.isEmpty()))
     .apply(Count.<String>perElement())
     .apply(MapElements
         .into(TypeDescriptors.strings())
         .via((KV<String, Long> wordCount) -> wordCount.getKey() + ": " + wordCount.getValue()))
     // CHANGE 3/3: The Google Cloud Storage path is required for outputting the results to.
     .apply(TextIO.write().to("gs://YOUR_OUTPUT_BUCKET/AND_OUTPUT_PREFIX"));

    p.run().waitUntilFinish();
  }
}

This example code 将让您更好地控制写入输出：

 /**
   * A {@link FilenamePolicy} produces a base file name for a write based on metadata about the data
   * being written. This always includes the shard number and the total number of shards. For
   * windowed writes, it also includes the window and pane index (a sequence number assigned to each
   * trigger firing).
   */
  protected static class PerWindowFiles extends FilenamePolicy {

    private final ResourceId prefix;

    public PerWindowFiles(ResourceId prefix) {
      this.prefix = prefix;
    }

    public String filenamePrefixForWindow(IntervalWindow window) {
      String filePrefix = prefix.isDirectory() ? "" : prefix.getFilename();
      return String.format(
          "%s-%s-%s", filePrefix, formatter.print(window.start()), formatter.print(window.end()));
    }

    @Override
    public ResourceId windowedFilename(int shardNumber,
                                       int numShards,
                                       BoundedWindow window,
                                       PaneInfo paneInfo,
                                       OutputFileHints outputFileHints) {
      IntervalWindow intervalWindow = (IntervalWindow) window;
      String filename =
          String.format(
              "%s-%s-of-%s%s",
              filenamePrefixForWindow(intervalWindow),
              shardNumber,
              numShards,
              outputFileHints.getSuggestedFilenameSuffix());
      return prefix.getCurrentDirectory().resolve(filename, StandardResolveOptions.RESOLVE_FILE);
    }

    @Override
    public ResourceId unwindowedFilename(
        int shardNumber, int numShards, OutputFileHints outputFileHints) {
      throw new UnsupportedOperationException("Unsupported.");
    }
  }

  @Override
  public PDone expand(PCollection<InputT> teamAndScore) {
    if (windowed) {
      teamAndScore
          .apply("ConvertToRow", ParDo.of(new BuildRowFn()))
          .apply(new WriteToText.WriteOneFilePerWindow(filenamePrefix));
    } else {
      teamAndScore
          .apply("ConvertToRow", ParDo.of(new BuildRowFn()))
          .apply(TextIO.write().to(filenamePrefix));
    }
    return PDone.in(teamAndScore.getPipeline());
  }

Answer 2

是的。每documentation of TextIO：

如果您想比默认策略更好地控制文件名的生成方式，也可以使用 TextIO.Write.to(FilenamePolicy) 设置自定义 FilenamePolicy

Answer 3

这是梁 2.1.0 的完全有效示例。您可以调用您的数据（例如 PCollection）

import org.apache.beam.sdk.io.FileBasedSink.FilenamePolicy;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.transforms.display.DisplayData;

@SuppressWarnings("serial")
public class FilePolicyExample {

    public static void main(String[] args) {
        FilenamePolicy policy = new WindowedFilenamePolicy("somePrefix");

        //data 
        data.apply(TextIO.write().to("your_DIRECTORY")
            .withFilenamePolicy(policy)
            .withWindowedWrites()
            .withNumShards(4));

    }

    private static class WindowedFilenamePolicy extends FilenamePolicy {

        final String outputFilePrefix;

        WindowedFilenamePolicy(String outputFilePrefix) {
            this.outputFilePrefix = outputFilePrefix;
        }

        @Override
        public ResourceId windowedFilename(
                ResourceId outputDirectory, WindowedContext input, String extension) {
            String filename = String.format(
                    "%s-%s-%s-of-%s-pane-%s%s%s",
                    outputFilePrefix,
                    input.getWindow(),
                    input.getShardNumber(),
                    input.getNumShards() - 1,
                    input.getPaneInfo().getIndex(),
                    input.getPaneInfo().isLast() ? "-final" : "",
                    extension);
            return outputDirectory.resolve(filename, StandardResolveOptions.RESOLVE_FILE);
        }

        @Override
        public ResourceId unwindowedFilename(
                ResourceId outputDirectory, Context input, String extension) {
            throw new UnsupportedOperationException("Expecting windowed outputs only");
        }

        @Override
        public void populateDisplayData(DisplayData.Builder builder) {
            builder.add(DisplayData.item("fileNamePrefix", outputFilePrefix)
                    .withLabel("File Name Prefix"));
        }
    }
}