p.run() 函数的 Apache Beam 问题

Question

我在我的一个 Apache Beam 脚本中遇到了问题。目前，它仍然是一个草稿，正在 DirectRunner 上运行。但我不断收到一个对我来说没有意义的错误。这是我的脚本的缩写版本：

options = PipelineOptions()
p = TestPipeline(options = options)
bill_info = (p
             | 'Create Pcollection' >> beam.Create('a')
             | 'Reformatting' >> beam.ParDo(ReformatAttributesFn())
             | 'Scrub First Name' >> beam.ParDo(ScrubFnameFn())
             | 'Scrub Last Name' >> beam.ParDo(ScrubLnameFn())
             | 'Fix Nickname' >> beam.Map(add_nickname, n_tbl=beam.pvalue.AsList(nickname_tbl_ex), return_n=False)
             | 'Check Sponsor' >> beam.Map(check_pol, p_tbl=beam.pvalue.AsList(pol_tbl_ex))
             | 'Check Bill' >> beam.Map(check_bill, b_tbl=beam.pvalue.AsList(bill_tbl_ex))
             | 'Final Formatting' >> beam.ParDo(FinalFormatFn())
             | 'Write To Text' >> beam.io.WriteToText('C:/Users/cmatt/Downloads/test_bill_votes', 
file_name_suffix='.csv'))
p.run()

我收到的错误是 p.run() 函数上的“TypeError：object() 没有参数”。但即使是最基本的管道，例如：

options = PipelineOptions()
p = TestPipeline(options = options)
bill_info = (p
             | 'Create Pcollection' >> beam.Create('a'))
p.run()

我仍然收到同样的错误。现在管道对象不应该接受任何输入；它只是一种通过管道运行所有 PCollection 的方法。但是我的脚本的哪一部分会导致错误？是不是我的虚拟环境中的某些模块没有正确排列？还是有一个我没有看到的单独问题？

感谢大家的帮助！

Answer 1

您的 python 脚本执行没有足够的上下文。您也收到与基本数据流指令相同的错误这一事实让我认为您的脚本有与 python 本身相关的错误，例如'mixed tabs and spaces'，你有这样的东西吗？

我只是从 python 命令行进行了一个快速测试，它工作了（步骤取自 Quickstart Using Python）：

>>> import apache_beam as beam
>>> from apache_beam.pipeline import PipelineOptions
>>> options = PipelineOptions()
>>> p = beam.Pipeline(options=options)
>>> bill_info = (p | 'Create Pcollection' >> beam.Create(['a','b']))
>>> p.run()
<apache_beam.runners.portability.fn_api_runner.RunnerResult object at 0x7fb57856d290>
>>>

如果以上信息没有帮助，您能否分享您正在执行的整个基本脚本以及您如何设置虚拟环境？这将有助于复制。