GCP Cloud Functions(Gen1)无法向BigQuery插入外汇JSON数据求助

1. 字段不匹配：直接用API返回的JSON结构插入，和BigQuery Schema的字段名、数量、类型完全对不上：
   • Schema里的timestamp是DATE类型，但API返回的是Unix时间戳（数字）；
   • Schema里的base_currency对应API里的base字段，exchange_date对应API里的date字段，未做字段映射；
   • Schema里的exchange_rates是重复型RECORD，但API返回的rates是单层字典，未转成字典列表格式。
2. 方法1报错：插入的字典字段数和Schema要求的4个字段不符，触发ValueError。
3. 方法2空表：数据格式完全不符合Schema，BigQuery会静默丢弃不符合要求的行，导致表为空。

步骤1：转换API数据为符合Schema的结构 #

对API返回的JSON做以下处理：

• 将Unix时间戳转换为DATE类型；
• 映射字段名到Schema定义的名称；
• 将rates字典转换为重复型RECORD的列表格式（每个货币对应一个字典元素）。

步骤2：正确调用insert_rows插入数据 #

确保传入的行数据是字典列表，每个字典的键和Schema字段名完全匹配，数据类型也对应。

完整可运行代码示例 #

import requests
import datetime
from google.cloud import bigquery

# 初始化BigQuery客户端
client = bigquery.Client()

def fetch_and_insert_exchange_rates(request):
    # 1. 调用API获取汇率数据
    api_url = "你的外汇API地址"
    response = requests.get(api_url)
    response.raise_for_status()
    raw_data = response.json()

    # 2. 转换数据为符合BigQuery Schema的格式
    # 转换Unix时间戳为DATE（UTC时区）
    timestamp_date = datetime.datetime.fromtimestamp(raw_data["timestamp"], datetime.timezone.utc).date()
    # 处理重复型RECORD字段
    exchange_rates_list = [
        {"exchange_currency": currency_code} 
        for currency_code in raw_data["rates"].keys()
        # 如需存储汇率值，需修改Schema添加rate字段，后续调整此处代码
    ]
    # 构造符合Schema的行数据
    row_data = [
        {
            "timestamp": timestamp_date,
            "base_currency": raw_data["base"],
            "exchange_date": raw_data["date"],
            "exchange_rates": exchange_rates_list
        }
    ]

    # 3. 插入BigQuery表
    table_id = "你的项目ID.数据集ID.表ID"
    table = client.get_table(table_id)
    errors = client.insert_rows(table, row_data)

    if not errors:
        return "数据插入成功", 200
    else:
        return f"插入失败: {errors}", 500

额外注意事项 #

• 如果需要存储汇率的具体数值，需修改Schema的exchange_rates字段，新增rate字段：

  bigquery.SchemaField("exchange_rates", "RECORD", mode="REPEATED",
                fields=[
                   bigquery.SchemaField("exchange_currency", "STRING", mode="NULLABLE"),
                   bigquery.SchemaField("rate", "FLOAT", mode="NULLABLE"), # 新增汇率字段
                ],
  )
  

  对应的转换代码同步调整：

  exchange_rates_list = [
      {"exchange_currency": currency_code, "rate": rate_value} 
      for currency_code, rate_value in raw_data["rates"].items()
  ]
  

• Cloud Functions需要绑定BigQuery数据插入权限（如roles/bigquery.dataEditor），否则会触发权限错误。
• 插入时若有错误，insert_rows会返回错误列表，务必打印或返回该信息以便排查问题。

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