Apache Beam DirectRunner无法与PubSub模拟器协同工作

1. 修正Beam参数与版本适配 #

不同版本的Beam Python SDK对PubSub模拟器的参数名存在差异：

• Beam 2.40+版本需使用--pubsub_emulator_host=127.0.0.1:8088，而非旧版的pubsubRootUrl/pubsub_root_url
• 代码中设置PipelineOptions时，需对应指定属性：

from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions

options = PipelineOptions()
gcp_options = options.view_as(GoogleCloudOptions)
gcp_options.project = 'test-project'
gcp_options.pubsub_emulator_host = '127.0.0.1:8088'
gcp_options.no_auth = True
options.view_as(StandardOptions).runner = 'DirectRunner'

2. 调整环境变量设置 #

CLOUDSDK_API_ENDPOINT_OVERRIDES_PUBSUB需包含完整HTTP前缀，正确设置命令：

export CLOUDSDK_API_ENDPOINT_OVERRIDES_PUBSUB=http://127.0.0.1:8088

设置后重启终端或重新加载环境变量，再运行Beam管道。

3. 确保模拟器环境变量生效 #

执行gcloud beta emulators pubsub env-init后，会输出以下变量，需手动执行export命令：

export PUBSUB_EMULATOR_HOST=127.0.0.1:8088
export PUBSUB_PROJECT_ID=test-project

也可将上述命令写入shell配置文件（如~/.bashrc、~/.zshrc）实现永久生效，Beam DirectRunner会优先读取PUBSUB_EMULATOR_HOST连接模拟器。

4. 清理认证冲突 #

• 清除残留默认凭据：

gcloud auth revoke --all

• 取消凭据环境变量：

unset GOOGLE_APPLICATION_CREDENTIALS

• 代码中明确禁用认证：

gcp_options.credentials = None

5. 前置验证模拟器连接 #

运行Beam管道前，用python-pubsub代码确认模拟器内主题、订阅已正确创建：

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()

topic_path = publisher.topic_path('test-project', 'your-topic')
subscription_path = subscriber.subscription_path('test-project', 'your-subscription')

# 检查主题
try:
    publisher.get_topic(request={"topic": topic_path})
    print("主题已存在")
except Exception as e:
    print(f"主题不存在: {e}")

# 检查订阅
try:
    subscriber.get_subscription(request={"subscription": subscription_path})
    print("订阅已存在")
except Exception as e:
    print(f"订阅不存在: {e}")

内容的提问来源于stack exchange，提问作者AMMJ