GCP：通过Python编写的Cloud Function向BigQuery插入数据失败

1. 缺失必要的导入依赖 #

代码中使用了bigquery.Client和functions_framework.http装饰器，但开头未导入对应模块，会直接导致运行报错。

修复：在代码开头添加导入语句：

import functions_framework
from google.cloud import bigquery

2. 未替换占位符表ID #

代码中的table_id = "<table_id>"是占位符，需替换为实际BigQuery表ID，格式为项目ID.数据集ID.表名。

3. Cloud Function服务账号权限不足 #

Cloud Function默认使用的服务账号需要具备BigQuery数据写入权限，否则会被拒绝访问。

修复：

• 找到当前Cloud Function使用的服务账号（通常为项目ID@appspot.gserviceaccount.com）
• 给该账号授予BigQuery 数据编辑器（roles/bigquery.dataEditor）角色，或更精细的bigquery.tables.insert权限

4. 请求JSON数据与表结构不匹配 #

插入的JSON字段名必须和BigQuery表字段名完全一致（大小写敏感），且数据类型要对应（比如表字段是INT64，不能传入字符串类型的数字）。

例如表结构为name:STRING, age:INT64，请求JSON需为{"name": "张三", "age": 25}，不能写成{"Name": "张三"}或{"age": "25"}。

5. 日志记录与异常处理不完善 #

原代码用print输出日志，且未捕获异常，导致无法定位具体失败原因。

修复：修改saveResponse函数，改用logging模块记录日志，并添加异常捕获：

def saveResponse(data):
    logging.info("Starting to insert data into BigQuery")
    client = bigquery.Client()
    table_id = "你的项目ID.数据集ID.表名"  # 替换为实际表ID
    try:
        table = client.get_table(table_id)
        rows_to_insert = [data]
        errors = client.insert_rows(table, rows_to_insert) 
        if errors == []:
            logging.info("Record created successfully")
            return None
        else:
            logging.error("Encountered errors while inserting rows: {}".format(errors))
            return errors
    except Exception as e:
        logging.error("Error during BigQuery operation: {}".format(str(e)))
        return str(e)

完整修复后的代码示例 #

import logging
import functions_framework
from google.cloud import bigquery

def saveResponse(data):
    logging.info("Starting to insert data into BigQuery")
    client = bigquery.Client()
    table_id = "你的项目ID.数据集ID.表名"  # 替换为实际表ID
    try:
        table = client.get_table(table_id)
        rows_to_insert = [data]
        errors = client.insert_rows(table, rows_to_insert) 
        if errors == []:
            logging.info("Record created successfully")
            return None
        else:
            logging.error("Encountered errors while inserting rows: {}".format(errors))
            return errors
    except Exception as e:
        logging.error("Error during BigQuery operation: {}".format(str(e)))
        return str(e)

@functions_framework.http
def create_http(request):
    # Set CORS headers for the preflight request
    if request.method == 'OPTIONS':
        headers = {
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Methods': '*',
            'Access-Control-Allow-Headers': 'Content-Type',
            'Access-Control-Max-Age': '3600'
        }
        return ('', 204, headers)

    # Set CORS headers for the main request
    headers = {
        'Content-Type':'application/json',
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Headers': 'Content-Type',
    }

    if request.method == 'POST':
        content_type = request.headers.get('Content-Type')
        if content_type == 'application/json':
            request_json = request.get_json(silent=True)
            if not request_json:
                logging.warning("No JSON data received in request")
                return ("Invalid JSON data", 400, headers)
            insert_errors = saveResponse(request_json)
            if insert_errors:
                return ("Failed to insert record: {}".format(insert_errors), 500, headers)
            return (request_json, 200, headers)
    return ("Unsupported request method", 405, headers)

排查步骤建议 #

1. 查看Cloud Function的日志页面，确认是否有导入错误、权限错误或数据格式错误的日志信息
2. 验证请求的JSON数据结构是否与BigQuery表字段完全匹配
3. 确认服务账号的BigQuery权限已正确配置

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