向GCS存储桶上传CSV触发Cloud Function后BigQuery插入空值问题

我明白你现在卡在哪了：把CSV上传到GCS桶时，Cloud Function确实触发了，但BigQuery里却只插入了空值。看日志就能找到根因——Pub/Sub传给函数的是GCS对象的元数据，不是CSV文件里的实际内容，你原来的代码直接把这份元数据往BQ里插，自然和BQ表的字段不匹配，导致空值。

那咱们来修正这个问题：需要在Cloud Function里多做两步，先根据元数据里的GCS文件路径读取CSV内容，再解析成BQ表对应的行数据，最后插入到BigQuery。下面是具体的步骤和修改后的代码：

第一步：给Cloud Function配置GCS读取权限 #

首先得让你的Cloud Function能读取GCS里的文件，给它的服务账号加上storage.objectViewer权限，用gcloud命令就能搞定：

gcloud projects add-iam-policy-binding YOUR_PROJECT_ID \
  --member serviceAccount:YOUR_FUNCTION_SERVICE_ACCOUNT \
  --role roles/storage.objectViewer

（记得把YOUR_PROJECT_ID和YOUR_FUNCTION_SERVICE_ACCOUNT换成你自己的项目ID和函数服务账号）

第二步：修改Cloud Function代码 #

替换你原来的Python代码，下面的版本加了读取GCS CSV、解析内容的逻辑，注释也标清楚了每一步做什么：

from google.cloud import bigquery, storage
import base64, json, sys, os, csv, io

def pubsub_to_bigquery(event, context):
    print("event:", event)
    print("context:", context)
    
    # 解析Pub/Sub传来的GCS对象元数据
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    gcs_metadata = json.loads(pubsub_message)
    print("GCS object metadata:", gcs_metadata)
    
    # 从元数据里提取桶名和文件名
    bucket_name = gcs_metadata['bucket']
    file_name = gcs_metadata['name']
    
    # 读取GCS里的CSV文件内容
    csv_content = read_csv_from_gcs(bucket_name, file_name)
    
    # 把CSV内容转换成BQ能接收的行数据格式
    rows_to_insert = parse_csv_to_rows(csv_content)
    
    # 插入到BigQuery表中
    write_to_bigquery(os.environ['my_dataset'], os.environ['my_table'], rows_to_insert)

def read_csv_from_gcs(bucket_name, file_name):
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    # 把文件内容下载成文本格式
    return blob.download_as_text()

def parse_csv_to_rows(csv_content):
    # 假设你的CSV文件带表头，用DictReader自动匹配字段
    csv_reader = csv.DictReader(io.StringIO(csv_content))
    # 把每一行转成字典，直接对应BQ表的字段
    return list(csv_reader)

def write_to_bigquery(dataset, table, rows):
    bigquery_client = bigquery.Client()
    dataset_ref = bigquery_client.dataset(dataset)
    table_ref = dataset_ref.table(table)
    table = bigquery_client.get_table(table_ref)
    
    errors = bigquery_client.insert_rows(table, rows)
    if errors != []:
        print(f"插入失败，错误信息: {errors}", file=sys.stderr)
    else:
        print(f"成功向BigQuery插入{len(rows)}条数据！")

第三步：重新部署Cloud Function #

用你原来的部署命令重新上传修改后的代码：

gcloud functions deploy pubsub_to_bigquery --region us-central1 --runtime python37 --trigger-topic data-test-notifications --source gs://pubsub_to_bigquery-bucket/test-code.zip

额外注意点 #

• 确保BigQuery表的字段名和CSV的表头完全一致，这样csv.DictReader解析出来的字典才能和BQ字段匹配。
• 如果你的CSV没有表头，就修改parse_csv_to_rows函数，手动指定BQ表的字段名，比如：

  def parse_csv_to_rows(csv_content):
      csv_reader = csv.reader(io.StringIO(csv_content))
      # 替换成你BQ表的实际字段名
      bq_fields = ['user_id', 'name', 'age']
      return [dict(zip(bq_fields, row)) for row in csv_reader]
  

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