Apache Beam Dataflow : 'NoneType'객체에 'parts'속성이 없습니다.

Question

저는 파이프 라인을 작성하여 pubsub에서 스트림을 읽고 apache 빔을 사용하여 Google Cloud Dataflow를 사용하여 bigquery에 씁니다. 이 코드가 있습니다

import apache_beam as beam 
from apache_beam.transforms.window import FixedWindows 

topic = 'projects/???/topics/???' 
table = '???.???' 

gcs_path = "gs://???" 

with beam.Pipeline(runner="DataflowRunner", argv=[ 
     "--project", "???", 
     "--staging_location", ("%s/staging_location" % gcs_path), 
     "--temp_location", ("%s/temp" % gcs_path), 
     "--output", ("%s/output" % gcs_path) 
    ]) as p: 
    (p 
    | 'winderow' >> beam.WindowInto(FixedWindows(60)) 
    | 'hello' >> beam.io.gcp.pubsub.ReadStringsFromPubSub(topic) 
    | 'hello2' >> beam.io.Write(beam.io.gcp.bigquery.BigQuerySink(table)) 
    ) 
    p.run().wait_until_finish() 

을하지만 그것을 실행이 오류 받고 있어요 :

No handlers could be found for logger "oauth2client.contrib.multistore_file" 
ERROR:root:Error while visiting winderow 
Traceback (most recent call last): 
    File ".\main.py", line 20, in <module> 
    p.run().wait_until_finish() 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\pipeline.py", line 339, in run 
    return self.runner.run(self) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\runners\dataflow\dataflow_runner.py", line 296, in run 
    super(DataflowRunner, self).run(pipeline) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\runners\runner.py", line 138, in run 
    pipeline.visit(RunVisitor(self)) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\pipeline.py", line 367, in visit 
    self._root_transform().visit(visitor, self, visited) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\pipeline.py", line 710, in visit 
    part.visit(visitor, pipeline, visited) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\pipeline.py", line 713, in visit 
    visitor.visit_transform(self) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\runners\runner.py", line 133, in visit_transform 
    self.runner.run_transform(transform_node) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\runners\runner.py", line 176, in run_transform 
    return m(transform_node) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\runners\dataflow\dataflow_runner.py", line 526, in run_ParDo 
    input_step = self._cache.get_pvalue(transform_node.inputs[0]) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\runners\runner.py", line 252, in get_pvalue 
    self._ensure_pvalue_has_real_producer(pvalue) 
    File "C:\ProgramData\Anaconda2\lib\site-packages\apache_beam\runners\runner.py", line 226, in _ensure_pvalue_has_real_producer 
    while real_producer.parts: 
AttributeError: 'NoneType' object has no attribute 'parts' 

이 코드 나 구성에 문제가 있습니까? 어떻게 작동시킬 수 있습니까?

Answer 1

아직 창 파이프 라인에 대한 경험이 없지만 개념에서 이해할 수 있도록 창은 파이프 라인 설정이 아닌 입력 데이터에 적용될 예정입니다.

것은이 사건되고, 아마 당신의 코드는 다음과 같아야합니다

with beam.Pipeline(runner="DataflowRunner", argv=[ 
     "--project", "???", 
     "--staging_location", ("%s/staging_location" % gcs_path), 
     "--temp_location", ("%s/temp" % gcs_path), 
     "--output", ("%s/output" % gcs_path) 
    ]) as p: 
    (p 
    | 'hello' >> beam.io.gcp.pubsub.ReadStringsFromPubSub(topic) 
    | 'winderow' >> beam.WindowInto(FixedWindows(60)) 
    | 'hello2' >> beam.io.Write(beam.io.gcp.bigquery.BigQuerySink(table)) 
    ) 
    p.run().wait_until_finish() 

공식 REPO뿐만 아니라 윈도우 작업에 대한 몇 가지 samples 있습니다.