apache-beam - 如何写入在运行时定义的文件名？

Question

我想写入一个 gs 文件，但在编译时我不知道文件名。它的名称基于在运行时定义的行为。我该如何继续？

Answer 1

如果你使用 Beam Java，你可以使用 FileIO.writeDynamic()为此(从目前正在发布的 Beam 2.3 开始 - 但您已经可以通过版本 2.3.0-SNAPSHOT 使用它)，或较旧的 DynamicDestinations API(在 Beam 2.2 中可用)。

使用示例 FileIO.writeDynamic()写一个 PCollection银行交易到 GCS 上不同路径的数量取决于交易的类型:

PCollection<BankTransaction> transactions = ...;
transactions.apply(
    FileIO.<BankTransaction, TransactionType>writeDynamic()
      .by(Transaction::getType)
      .via(BankTransaction::toString, TextIO.sink())
      .to("gs://bucket/myfolder/")
      .withNaming(type -> defaultNaming("transactions_", ".txt"));

以 DynamicDestinations 为例使用，见 example code in the TextIO unit tests .

或者，如果您想将每条记录写入自己的文件，只需使用 FileSystems API(特别是 FileSystems.create() )来自 DoFn .