Apache Beam解析BigQuery JSON Schema报错：找不到TableFieldSchema属性

问题背景 #

我尝试从包含嵌套和重复列的BigQuery表结构JSON文件中解析Schema，使用代码如下：

import apache_beam as beam
from apache_beam.io.gcp.bigquery import parse_table_schema_from_json
import json

schema_data = json.dumps(json.load(open("/content/sample_data/schema.json")))
table_schema = parse_table_schema_from_json(schema_data)
print(table_schema)

运行时抛出错误：

module 'apache_beam.io.gcp.internal.clients.bigquery' has no attribute 'TableFieldSchema'

已通过pip install apache-beam[gcp]安装GCP专属的Apache Beam库，Schema内容如下：

{
  "fields": [
    {
      "fields": [
        {
          "mode": "NULLABLE",
          "name": "second",
          "type": "INTEGER"
        },
        {
          "mode": "NULLABLE",
          "name": "minute",
          "type": "INTEGER"
        },
        {
          "mode": "NULLABLE",
          "name": "hour",
          "type": "INTEGER"
        },
        {
          "mode": "NULLABLE",
          "name": "timeZoneId",
          "type": "STRING"
        },
        {
          "fields": [
            {
              "mode": "NULLABLE",
              "name": "month",
              "type": "INTEGER"
            },
            {
              "mode": "NULLABLE",
              "name": "day",
              "type": "INTEGER"
            },
            {
              "mode": "NULLABLE",
              "name": "year",
              "type": "INTEGER"
            }
          ],
          "mode": "NULLABLE",
          "name": "date",
          "type": "RECORD"
        }
      ],
      "mode": "NULLABLE",
      "name": "lastModifiedDateTime",
      "type": "RECORD"
    },
    {
      "mode": "REPEATED",
      "name": "companionCreativeIds",
      "type": "INTEGER"
    },
    {
      "mode": "NULLABLE",
      "name": "masterCreativeId",
      "type": "INTEGER"
    },
    {
      "mode": "NULLABLE",
      "name": "name",
      "type": "STRING"
    },
    {
      "mode": "NULLABLE",
      "name": "id",
      "type": "INTEGER"
    },
    {
      "fields": [
        {
          "mode": "NULLABLE",
          "name": "name",
          "type": "STRING"
        },
        {
          "mode": "NULLABLE",
          "name": "value",
          "type": "STRING"
        }
      ],
      "mode": "REPEATED",
      "name": "soft_error_fields",
      "type": "RECORD"
    },
    {
      "mode": "NULLABLE",
      "name": "dw_ingest_time",
      "type": "TIMESTAMP"
    },
    {
      "mode": "NULLABLE",
      "name": "dw_partition_date",
      "type": "DATE"
    },
    {
      "mode": "NULLABLE",
      "name": "dw_publish_time",
      "type": "TIMESTAMP"
    },
    {
      "mode": "NULLABLE",
      "name": "dw_source_object_name",
      "type": "STRING"
    },
    {
      "mode": "NULLABLE",
      "name": "dw_batch_id",
      "type": "STRING"
    }
  ]
}

解决方案 #

1. 修正Schema解析逻辑（最直接） #

原代码中json.dumps(json.load(...))属于多余操作，会将加载后的字典二次序列化为字符串，可能引发解析异常。直接读取文件原始字符串传入解析函数即可：

import apache_beam as beam
from apache_beam.io.gcp.bigquery import parse_table_schema_from_json

# 直接读取文件内容为原始JSON字符串
with open("/content/sample_data/schema.json", "r") as f:
    schema_str = f.read()

table_schema = parse_table_schema_from_json(schema_str)
print(table_schema)

2. 升级Apache Beam版本 #

该错误多因Beam版本迭代导致内部BigQuery客户端API变更，TableFieldSchema在新版本中已调整导入路径或实现方式。执行以下命令升级到稳定版本：

pip install --upgrade apache-beam[gcp] --no-cache-dir

3. 手动构建TableSchema（自定义场景） #

如果需要更灵活的Schema控制，可手动通过TableSchema和TableFieldSchema类构建：

import apache_beam as beam
from apache_beam.io.gcp.bigquery import TableSchema, TableFieldSchema
import json

def build_field_schema(field_dict):
    field = TableFieldSchema()
    field.name = field_dict["name"]
    field.type = field_dict["type"]
    field.mode = field_dict["mode"]
    if "fields" in field_dict:
        field.fields = [build_field_schema(f) for f in field_dict["fields"]]
    return field

with open("/content/sample_data/schema.json", "r") as f:
    schema_dict = json.load(f)

table_schema = TableSchema(fields=[build_field_schema(f) for f in schema_dict["fields"]])
print(table_schema)

4. 重新安装依赖确保完整性 #

若依赖安装不完整，卸载后重新安装：

pip uninstall -y apache-beam apache-beam[gcp]
pip install apache-beam[gcp]

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