具有 apache beam 和运行时参数的动态路径

Question

我正在创建一个管道模板，它接受一些输入文件并计算其上的单词数。到目前为止一切正常，但问题是我需要传递另一个参数（来自我调用模板的函数），该参数让我传递文件名，以便我可以使用它创建路径。

我将向您展示我想要的示例，尽管我知道管道无法在管道构建期间或运行时上下文之外访问运行时参数，这可以帮助您了解我需要做什么：

class tempatableTest(PipelineOptions):
@classmethod
def _add_argparse_args(cls,parser):
    parser.add_value_provider_argument(
        '--input',
        type=str,
        help='path to the input file'
    )
    parser.add_value_provider_argument(
        '--fdinamic',
        type=str,
        help='folder name'
    )

templatable_test = PipelineOptions().view_as(tempatableTest)
beam_options= PipelineOptions()
input = templatable_test.input
dinamicName = templatable_test.fdinamic.get()

with beam.Pipeline(options=beam_options) as p:
    lines = p | beam.io.ReadFromText(input)
    len = lines | beam.combiners.Count.Globally()
    len | 'countTotalLen' >> beam.io.WriteToText(f'gs://bucket-test-out/processedFile/{dinamicName}/count.txt')

如果我使用 templatable_test.fdinamic .get() 我会收到运行时错误，但如果删除 .get() 我会在文件夹上得到一个超长的名称。

我知道这可能不是正确的方法，但只是为了说明我需要做什么，谢谢您的帮助。

Answer 1

不幸的是，WriteToText 转换不能用于此目的，因为它当前仅支持固定目标。因此，为了将文件写入动态目标，您需要使用 fileio 模块 支持动态目标。虽然这确实意味着切换到使用实验 WriteToFiles 转换。