问题描述
我目前有代码:
gs_folder = sys.argv[1]
options = PipelineOptions(
runner='DataflowRunner',project='xxx',job_name=xxx{uuid.uuid4()}',region='us-central1',temp_location='xxx')
gfs = gcs.GCSFileSystem(options)
p = beam.Pipeline(options=options)
discover_empty = p | 'Filenames' >> beam.Create(gfs.match([gs_folder])[0].Metadata_list) | \
'Reshuffle' >> beam.Reshuffle() | \
'Delete empty files' >> beam.ParDo(DeleteEmpty(gfs))
p.run()
WARNING:apache_beam.options.pipeline_options:discarding unparseable args: ['gs://xx/xx']
这没有多大意义,因为这是我要执行删除操作的文件夹。此外,看起来数据流作业确实成功运行,但应该删除的文件没有正确删除。我应该如何在此处传递管道选项 arg?
我也有一些关于此过程的后续问题。看起来 beam.Create()
在本地运行,然后切换到数据流。我怎样才能让那部分管道在数据流上运行?
解决方法
确保您将标志传递为 --input=gs//... 该错误看起来像是您的命令行调用无效,并且 gs 路径被解释为整个参数。
beam.Create 作为管道的一部分运行,但传递给它的参数是在本地计算的。要在管道中进行计算,请使用 beam.Create(None),然后使用运行匹配逻辑的 DoFn。
,你可以作为命令行参数传递
--db_port 5432 --db_pass XXXX
import argparse
# initializing Pipeline object - parse CL args
parser = argparse.ArgumentParser()
parser.add_argument('--db_pass')
parser.add_argument('--db_port')
parser.add_argument('--db_host')
args,beam_args = parser.parse_known_args()
# initializing Pipeline object
options = PipelineOptions(beam_args)
options.view_as(StandardOptions).streaming = True
p = beam.Pipeline(options=options)
args.db_pass // your value