Unable to read Pub/Sub messages with Apache Beam (Python SDK)

Question

I'm trying to stream messages from a Pub/Sub topic with the Beam programming framework (Python SDK) and write them out to the console.

This is my code (with apache-beam==2.27.0):

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

TOPIC_PATH = "projects/<project-id>/topics/<topic-id>"

def run(pubsub_topic):
    options = PipelineOptions(
        streaming=True
    )
    runner = 'DirectRunner'

    print("I reached before pipeline")

    with beam.Pipeline(runner, options=options) as pipeline:
        (
            pipeline
            | "Read from Pub/Sub topic" >> beam.io.ReadFromPubSub(topic=pubsub_topic)
            | "Writing to console" >> beam.Map(print)
        )

    print("I reached after pipeline")

    result = pipeline.run()
    result.wait_until_finish()


run(TOPIC_PATH)

When I execute this pipeline however, I get this TypeError:

ERROR:apache_beam.runners.direct.executor:Exception at bundle <apache_beam.runners.direct.bundle_factory._Bundle object at 0x1349763c0>, due to an exception.

TypeError: create_subscription() takes from 1 to 2 positional arguments but 3 were given

In the end it says:

ERROR:apache_beam.runners.direct.executor:Giving up after 4 attempts.

I'm not sure, what I'm doing wrong, thanks in advance for your help.

Answer 1

I don't know exactly where the error is but can you consider using one of the following Beam examples as a model and going from there ?

https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/gcp/pubsub_it_pipeline.py

https://github.com/apache/beam/blob/release-2.27.0/sdks/python/apache_beam/examples/snippets/snippets.py#L684

Answer 2

I had the same problem when I installed by "pip install apache-beam". When I switched to "pip install apache-beam [gcp]", it worked for me, even with DirectRunner.