在此文档(https://beam.apache.org/documentation/programming-guide/#additional-outputs)的4.5.2节中,有一个
看起来很难导入
我从这里的示例代码获取了所有代码:https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/multiple_output_pardo.py
我的run()方法的示例
"
pvalue.TaggedOutput()
被提及。看起来很难导入
pvalue
,我已经从Apache文档中复制了导入行,并在def run()
中使用了--save_main_session
选项以及save_main_session=True
,并在开始管道之前使用pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
。所有导入都对所有函数有效,所有类都可以在所有函数中使用,但是pvalue
无法识别。我也尝试了所有可能的组合以及省略它们,pvalue
始终未知。我从这里的示例代码获取了所有代码:https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/cookbook/multiple_output_pardo.py
尽管如此,没有p值。
NameError:名称'pvalue'未定义[在运行“generatedPtransform-1725”时]
仅当我使用Dataflowrunner时才会生成此错误,而不是使用Directrunner时。
我的DoFn示例
class Splitter(beam.DoFn):
TAG1 = 'kleintje'
TAG2 = 'grootje'
def process(self, element):
splittertid = element.get('id')
if splittertid < 100:
yield pvalue.TaggedOutput(self.TAG1, element)
else:
yield pvalue.TaggedOutput(self.TAG2, element)
我的run()方法的示例
def run(argv=None, save_main_session=True):
sources = [
json.loads('{"id":72234,"value":1'),
json.loads('{"id":23,"value":2}')
]
parser = argparse.ArgumentParser()
known_args, pipeline_args = parser.parse_known_args(argv)
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
with beam.Pipeline(options=pipeline_options) as p:
| beam.Create(sources)
| beam.ParDo(Splitter()).with_outputs(Splitter.TAG1,Splitter.TAG2,main=Splitter.TAG1)
"
**我的导入**
"from __future__ import absolute_import
import argparse
import logging
import re
import json
import datetime
from jsonschema import validate
import apache_beam as beam
from apache_beam import pvalue
from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.io.gcp.bigquery_tools import parse_table_schema_from_json
from apache_beam import pvalue
这一行吗?你也在使用这个吗? - undefined