使用Apache Beam Dataflow写入BigQuery遇TypeError及安装问题求助

一、写入BigQuery的TypeError问题 #

错误栈 #

| "Write to BigQuery" >> beam.io.WriteToBigQuery(
                             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\x\Python\Python311\Lib\site-packages\apache_beam\io\gcp\bigquery.py", line 2083, in __init__
    self.table_reference = bigquery_tools.parse_table_reference(
                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\x\Python\Python311\Lib\site-packages\apache_beam\io\gcp\bigquery_tools.py", line 263, in parse_table_reference
    if isinstance(table, TableReference):
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: isinstance() arg 2 must be a type, a tuple of types, or a union

环境信息 #

• Python版本：3.11
• Apache Beam版本：2.57.0
• google-cloud-bigquery版本：3.26.0
• 运行器：DirectRunner（本地测试）

已尝试操作 #

1. 确认版本兼容：根据Beam仓库配置，google-cloud-bigquery>=2.0.0,<4符合要求
2. table_id格式为project_id:dataset_id.table_name
3. 尝试两种Schema定义方式：
   • 字典格式：

     table_schema = {
             "fields": [
                 {"name": "trip_id", "type": "STRING"},
                 {"name": "vehicle_id", "type": "STRING"},
                 {"name": "customer_id", "type": "STRING"},
                 {"name": "driver_id", "type": "STRING"}
             ]
         }
     

   • 字符串格式：

     schema='trip_id: STRING, vehicle_id: STRING, customer_id: STRING, driver_id: STRING,',
     

4. 核心代码片段：

   # 创建Beam管道
   with beam.Pipeline(options=options) as p:
       (
           p
               | "Start" >> beam.Create([None]) 
               | "Read from Cloud SQL" >> beam.ParDo(ReadFromCloudSQL()) 
               | "Print results" >> beam.Map(print)  # 控制台打印结果
               | "Write to BigQuery" >> beam.io.WriteToBigQuery(
                   table=table_id,
                   schema=table_schema,
                   create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
                   write_disposition=beam.io.gcp.bigquery.BigQueryDisposition.WRITE_APPEND 
   
                )
       )
   

二、apache-beam[gcp]安装失败问题 #

安装错误信息 #

$ pip install apache-beam[gcp]
ERROR: Could not find a version that satisfies the requirement google-apitools<0.5.32,>=0.5.31; extra == "gcp" (from apache-beam[gcp]) (from versions: 0.5.32)
ERROR: No matching distribution found for google-apitools<0.5.32,>=0.5.31; extra == "gcp"

已尝试操作 #

1. 使用私有PyPI仓库，怀疑仓库限制导致问题
2. 清理pip缓存
3. 指定公共PyPI源安装：--index-url=https://pypi.org/simple，问题依旧

解决方案 #

针对TypeError问题 #

该错误源于google-cloud-bigquery 3.x版本中TableReference类的路径变更：从google.cloud.bigquery.table.TableReference迁移到google.cloud.bigquery.reference.TableReference，但Beam 2.57.0的代码仍在旧路径查找，导致TableReference不是有效类型。

解决方法二选一：

1. 降级google-cloud-bigquery到2.x稳定版本（如2.34.4），与Beam 2.57.0的依赖完全匹配
2. 升级Apache Beam到2.58.0及以上版本，这些版本已适配google-cloud-bigquery 3.x的类路径变更

针对apache-beam[gcp]安装失败问题 #

Beam 2.57.0的gcp extras依赖约束google-apitools>=0.5.31,<0.5.32与PyPI上仅有的0.5.32版本冲突，导致安装失败。

解决方法二选一：

1. 手动安装指定版本的依赖，跳过严格约束：

   pip install apache-beam==2.57.0 google-cloud-bigquery==2.34.4 google-apitools==0.5.32
   

2. 升级到Beam 2.58.0及以上版本，这些版本已将google-apitools的依赖约束放宽至<0.5.33，可正常安装

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