Python Dataflow从PubSub读数据写入BigQuery异常排查求助

问题背景 #

已创建基于Python的Dataflow流式作业，目标是从PubSub Topic读取数据写入BigQuery，但作业无法正常完成，怀疑PubSub读取环节存在问题。BigQuery表已成功创建，使用静态测试数据的管道可正常将数据写入BigQuery，但原作业执行异常。

原作业代码 #

import os
import json 
import logging
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions
from dotenv import load_dotenv
from create_dataset_and_table_in_bq import create_bq_dataset_and_table
load_dotenv()

current_dir = os.getcwd()
gcp_credential_path = os.getenv('gcp_credential_path', '')
project_id = os.getenv('project_id', '')
topic_name = os.getenv('topic_name', '')
region = os.getenv('region', '')
bucket_path = os.getenv('bucket_path', '')
job_name = os.getenv('job_name', '')
dataset_id = os.getenv('dataset_id', '')
table_id = os.getenv('table_id', '')

# Get the root of the current drive
root_dir = os.path.splitdrive(current_dir)[0] + os.sep
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = os.path.join(root_dir, gcp_credential_path)
 
options = PipelineOptions()
google_cloud_options = options.view_as(GoogleCloudOptions)
google_cloud_options.project = project_id
google_cloud_options.region = region
google_cloud_options.job_name = job_name 
google_cloud_options.staging_location = f"{bucket_path}/staging"
google_cloud_options.temp_location = f"{bucket_path}/temp"
options.view_as(StandardOptions).runner = 'DataflowRunner'
options.view_as(StandardOptions).streaming = True

def transform_data(element):
    print(f"Received message: {element}")
    return json.loads(element.decode('utf-8'))

dataset_created, table_created = create_bq_dataset_and_table(region, dataset_id, table_id)

# BigQuery schema
schema = ('primary_title:STRING, num_votes:INTEGER') 

subscription_name = f'projects/{project_id}/subscriptions/dataflow-insight-hub-pubsub-topic-2-sub'


with beam.Pipeline(options=options) as p:
    (p
    | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(topic=f'projects/{project_id}/topics/{topic_name}', with_attributes=True)
    | 'Transform Data' >> beam.Map(transform_data)
    | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
            table=f"{project_id}:{dataset_id}.{table_id}",
            schema=schema,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
        )
    )

测试管道代码 #

with beam.Pipeline(options=options) as p:
        messages = p | 'Create Test Data' >> beam.Create([
            b'{"primary_title": "Test Movie", "num_votes": 1000}',
            b'{"primary_title": "Another Movie", "num_votes": 500}'
        ])
        (messages
        | 'Log and Transform Data' >> beam.Map(log_and_transform_data)
        | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
                table=f"{project_id}:{dataset_id}.{table_id}",
                schema=schema,
                write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
            )
        )

排查方向与解决方案 #

1. 修正PubSub读取返回值的处理逻辑 #

原代码中ReadFromPubSub设置了with_attributes=True，此时返回的element是元组（message_data, attributes），而非单纯的字节数据。但transform_data函数直接对元组调用decode方法，会触发类型错误。

• 修复方式（保留属性的场景）：

  def transform_data(element):
      message_data, attributes = element
      print(f"Received message: {message_data}, attributes: {attributes}")
      return json.loads(message_data.decode('utf-8'))
  

• 若不需要属性，可直接移除with_attributes=True参数，此时element为原始字节数据，原transform_data逻辑可正常使用。

2. 确认PubSub权限配置 #

• 确保Dataflow作业使用的服务账号（密钥对应的账号或默认Dataflow服务账号）拥有roles/pubsub.subscriber权限，能正常订阅目标Topic。
• 检查Topic是否有消息流入：通过GCP PubSub控制台查看Topic的消息统计，确认是否有数据可被读取。

3. 查看Dataflow作业日志定位报错 #

进入GCP控制台的Dataflow作业详情页，查看Worker日志和Job日志，搜索Error、Failed、PubSub等关键词，获取具体报错信息（如权限不足、数据格式错误等）。

4. 验证订阅的使用 #

原代码定义了subscription_name但未使用，当通过topic参数读取时，Dataflow会自动创建临时订阅。若Topic存在大量未确认消息，可能影响作业启动：

• 可显式指定已创建的订阅：

  | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(subscription=subscription_name)
  

• 确保该订阅未被其他消费者占用，且权限配置正确。

5. 确认流式作业状态 #

流式作业默认持续运行，不会自动结束。若作业处于Running状态但无数据写入，大概率是Topic无消息或上述格式/权限问题导致数据无法处理；若作业处于Failed状态，日志会明确给出失败原因。

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