Python中无法从Pub/Sub向BigQuery写入数据问题排查

该Pipeline卡住是由三个核心问题叠加导致的：消息序列化格式错误、字段引用不存在的Key、BigQuery Schema与数据不匹配，以下是具体问题和修复方法：

一、核心问题排查 #

1. publishEvents.py的消息序列化错误 #

你在发布消息时，将json.loads(line)得到的字典直接转为字符串（data=str(traffic)），发送的是Python字典的字符串表示（键为单引号），而非标准JSON格式。例如发送内容为：

{'user_id': '102105290400257936', ...}

这会导致parse_json_message中的json.loads(message)抛出解析错误，Pipeline因未捕获异常卡住。

修复方法：使用json.dumps()将字典序列化为标准JSON字符串：

# 替换原代码中的data=str(traffic)
data = json.dumps(traffic)

2. parse_json_message引用不存在的字段 #

输入数据（Input_Data.json）中没有create_ts、ifa、ifv、chip_balance字段，但parse_json_message直接读取这些Key，会触发KeyError，导致Pipeline处理失败。

修复方法：

• 对需要新增的字段（如create_ts）生成对应值（比如当前处理时间）
• 对输入中不存在的可选字段，使用row.get('key', None)避免KeyError

修改后的parse_json_message示例：

def parse_json_message(message):
    row = json.loads(message)
    return {
        'user_id': row['user_id'],
        'event_ts': row['event_ts'],
        'create_ts': datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f UTC'),
        'event_id': row['event_id'],
        'ifa': row.get('ifa'),
        'ifv': row.get('ifv'),
        'country': row['country'],
        'chip_balance': row.get('chip_balance'),
        'game': row['game'],
        'user_group': row['user_group'],
        'user_condition': row['user_condition'],
        'device_type': row['device_type'],
        'device_model': row['device_model'],
        'user_name': row['user_name'],
        'fb_connect': row['fb_connect'],
        'is_active_event': row['is_active_event'],
        'event_payload': row['event_payload']
    }

3. BigQuery Schema的REQUIRED字段不匹配 #

BigQuery Schema中create_ts被设为REQUIRED类型，但原输入数据无此字段，且代码未生成对应值，即使前面问题解决，写入BigQuery时也会因缺少必填字段失败。

修复方法：

• 按上述方式在parse_json_message中生成create_ts的值
• 若业务允许，将create_ts的mode改为NULLABLE

二、修复后的完整代码 #

修复后的publishEvents.py #

import json
from google.cloud import pubsub_v1
from concurrent import futures
from typing import Callable

project='nectar-259905'
topic='incomingData'
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project, topic)
publish_futures = []

def get_callback(
    publish_future: pubsub_v1.publisher.futures.Future, data: str
) -> Callable[[pubsub_v1.publisher.futures.Future], None]:
    def callback(publish_future: pubsub_v1.publisher.futures.Future) -> None:
        try:
            print(publish_future.result(timeout=60))
        except futures.TimeoutError:
            print(f"Publishing {data} timed out.")

    return callback

# 若Input_Data.json是单条JSON对象，直接加载整个文件
with open('/home/test/Resources/Input/Input_Data.json', 'r') as f:
    traffic = json.load(f)
    data = json.dumps(traffic)  # 生成标准JSON字符串
    publish_future = publisher.publish(topic_path, data.encode("utf-8"))
    publish_future.add_done_callback(get_callback(publish_future, data))
    publish_futures.append(publish_future)
futures.wait(publish_futures, return_when=futures.ALL_COMPLETED)

print(f"Published messages with error handler to {topic_path}.")

注：若Input_Data.json是多行JSON Lines格式（每行一个JSON对象），保留原for循环，但需确保每行是合法JSON，且用json.dumps()序列化。

修复后的readFromPubSub.py #

import argparse
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
from datetime import datetime
import json

def parse_json_message(message):
    row = json.loads(message)
    return {
        'user_id': row['user_id'],
        'event_ts': row['event_ts'],
        'create_ts': datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f UTC'),
        'event_id': row['event_id'],
        'ifa': row.get('ifa'),
        'ifv': row.get('ifv'),
        'country': row['country'],
        'chip_balance': row.get('chip_balance'),
        'game': row['game'],
        'user_group': row['user_group'],
        'user_condition': row['user_condition'],
        'device_type': row['device_type'],
        'device_model': row['device_model'],
        'user_name': row['user_name'],
        'fb_connect': row['fb_connect'],
        'is_active_event': row['is_active_event'],
        'event_payload': row['event_payload']
    }

def run(argv=None, save_main_session=True):
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--subscription',
        dest='subscription',
        default='projects/nectar-259905/subscriptions/bq_subscribe',
        help='Input Pub/Sub subscription')

    parser.add_argument(
        '--table_spec',
        dest='table_spec',
        default='nectar-259905:JSONData.bq_table',
        help='Destination BigQuery table.')

    known_args, pipeline_args = parser.parse_known_args(argv)
    options = PipelineOptions(pipeline_args)
    options.view_as(StandardOptions).streaming = True

    with beam.Pipeline(options=options) as p:
        table_schema = {
            'fields': [
                {"name": "user_id", "mode": "REQUIRED", "type": "STRING"},
                {"name": "event_ts", "mode": "REQUIRED", "type": "STRING"},
                {"name": "create_ts", "mode": "REQUIRED", "type": "STRING"},
                {"name": "event_id", "mode": "REQUIRED", "type": "STRING"},
                {"name": "ifa", "mode": "NULLABLE", "type": "STRING"},
                {"name": "ifv", "mode": "NULLABLE", "type": "STRING"},
                {"name": "country", "mode": "NULLABLE", "type": "STRING"},
                {"name": "chip_balance", "mode": "NULLABLE", "type": "STRING"},
                {"name": "game", "mode": "NULLABLE", "type": "STRING"},
                {"name": "user_group", "mode": "NULLABLE", "type": "STRING"},
                {"name": "user_condition", "mode": "NULLABLE", "type": "STRING"},
                {"name": "device_type", "mode": "NULLABLE", "type": "STRING"},
                {"name": "device_model", "mode": "REQUIRED", "type": "STRING"},
                {"name": "user_name", "mode": "REQUIRED", "type": "STRING"},
                {"name": "fb_connect", "mode": "NULLABLE", "type": "BOOLEAN"},
                {"name": "is_active_event", "mode": "NULLABLE", "type": "BOOLEAN"},
                {"name": "event_payload", "mode": "NULLABLE", "type": "STRING"}
            ]
        }

        (p
            | 'Read from pubsub' >> beam.io.ReadFromPubSub(subscription=known_args.subscription)
            | 'Parse JSON messages' >> beam.Map(parse_json_message)
            | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
                known_args.table_spec,
                schema=table_schema,
                write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
                create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED))

if __name__ == '__main__':
    run()

三、额外排查建议 #

• 添加日志：在parse_json_message中加入logging.info，或用beam.Map(lambda x: (logging.info(x), x)[1])步骤打印消息，方便定位错误
• 检查Pub/Sub订阅：确认订阅正常接收消息，无未确认消息堆积
• 验证权限：确保运行Pipeline的账号拥有BigQuery写入权限

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