如何每周自动将Google Drive中CSV导入BigQuery且不覆盖历史数据？

前置准备 #

• 把所有待加载的CSV统一放在同一个Google Drive文件夹里，确保所有CSV的列名、数据类型完全一致
• 拥有Google Cloud平台的以下权限：BigQuery数据编辑权限、Cloud Functions部署权限、Cloud Scheduler权限，以及Google Drive文件读取权限

步骤1：创建BigQuery目标表 #

如果还没有目标表，按以下操作创建：

1. 打开BigQuery控制台，进入你的数据集
2. 点击「创建表」，来源选「空表」
3. 在「架构」里手动添加CSV对应的列（列名、数据类型要和CSV完全匹配）
4. 也可以直接用SQL语句创建（替换成你的列信息）：

CREATE TABLE `你的GCP项目ID.数据集ID.目标表名` (
  用户ID STRING,
  交易金额 INT64,
  交易日期 DATE
  -- 按你的CSV列依次定义
)

步骤2：建立BigQuery与Google Drive的连接 #

让BigQuery能读取Drive里的文件：

1. 在BigQuery控制台的数据集页面，点击「创建连接」
2. 连接类型选择「Google Drive」，设置一个连接名称（比如drive-csv-link）
3. 按提示授权BigQuery访问你的Google Drive，确保能访问存放CSV的目标文件夹

步骤3：部署Cloud Function实现自动加载逻辑 #

我们用Python写一个简单的函数，自动识别Drive里的新CSV并追加到BigQuery：

函数逻辑说明 #

• 记录已加载的文件，避免重复导入
• 遍历Drive目标文件夹的CSV文件
• 对未加载的文件，用追加模式导入到BigQuery
• 标记文件为已加载（可选：将已加载文件移至归档文件夹）

部署操作 #

1. 打开Cloud Functions控制台，点击「创建函数」
2. 函数名设为load-drive-csv-to-bq，运行时选择Python 3.11（或更高版本）
3. 触发器选择「Cloud Pub/Sub」，新建一个主题（比如drive-csv-load-trigger）
4. 替换下面的示例代码中的参数（Drive文件夹ID、GCP项目ID等），粘贴到「main.py」里：

from google.cloud import bigquery
from googleapiclient.discovery import build
from google.oauth2 import service_account

# 初始化客户端
bq_client = bigquery.Client()
drive_service = build('drive', 'v3', credentials=service_account.Credentials.from_service_account_file('service-key.json'))

# 配置参数（替换成你的信息）
DRIVE_FOLDER_ID = '你的Drive文件夹ID'  # 从文件夹URL末尾获取
BQ_PROJECT_ID = '你的GCP项目ID'
BQ_DATASET_ID = '你的数据集ID'
BQ_TABLE_ID = '你的目标表名'
BQ_LOG_TABLE = f'{BQ_PROJECT_ID}.{BQ_DATASET_ID}.loaded_files_log'

def load_new_csv(event, context):
    # 创建加载日志表（记录已加载的文件）
    bq_client.query(f"""
        CREATE TABLE IF NOT EXISTS {BQ_LOG_TABLE} (
            file_id STRING,
            file_name STRING,
            loaded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
        )
    """).result()

    # 获取已加载的文件ID列表
    loaded_file_ids = [row.file_id for row in bq_client.query(f"SELECT file_id FROM {BQ_LOG_TABLE}").result()]

    # 列出Drive文件夹中的CSV文件
    csv_files = drive_service.files().list(
        q=f"'{DRIVE_FOLDER_ID}' in parents and mimeType='text/csv' and trashed=false",
        fields="files(id, name)"
    ).execute().get('files', [])

    # 处理未加载的文件
    loaded_count = 0
    for file in csv_files:
        if file['id'] not in loaded_file_ids:
            # 配置BigQuery加载规则：追加模式、跳过表头
            job_config = bigquery.LoadJobConfig(
                source_format=bigquery.SourceFormat.CSV,
                skip_leading_rows=1,  # CSV有表头就设1，没有设0
                write_disposition=bigquery.WriteDisposition.WRITE_APPEND,
                schema=bq_client.get_table(f"{BQ_PROJECT_ID}.{BQ_DATASET_ID}.{BQ_TABLE_ID}").schema
            )

            # Drive文件的BigQuery访问路径
            file_uri = f"gs://drive-{BQ_PROJECT_ID}-{DRIVE_FOLDER_ID}/{file['id']}"

            # 执行加载任务
            load_job = bq_client.load_table_from_uri(
                file_uri,
                f"{BQ_PROJECT_ID}.{BQ_DATASET_ID}.{BQ_TABLE_ID}",
                job_config=job_config
            )
            load_job.result()  # 等待加载完成

            # 记录到日志表
            bq_client.query(f"""
                INSERT INTO {BQ_LOG_TABLE} (file_id, file_name)
                VALUES ('{file['id']}', '{file['name']}')
            """).result()

            # 可选：将已加载文件移至归档文件夹
            # drive_service.files().update(
            #     fileId=file['id'],
            #     addParents='你的归档文件夹ID',
            #     removeParents=DRIVE_FOLDER_ID
            # ).execute()

            loaded_count += 1

    return f"完成加载，共处理{loaded_count}个新CSV文件"

5. 上传你的GCP服务账号密钥文件（命名为service-key.json），确保该账号有Drive读取和BigQuery写入权限
6. 点击「部署」，等待函数部署完成

步骤4：设置每周自动触发 #

用Cloud Scheduler实现每周定时执行：

1. 打开Cloud Scheduler控制台，点击「创建作业」
2. 作业ID设为weekly-drive-csv-load，频率用Cron表达式设置（比如0 0 * * 1表示每周一凌晨0点执行，可按需调整）
3. 目标选择「Pub/Sub」，主题选之前创建的drive-csv-load-trigger，消息内容随便填（比如"start load"）
4. 选择你的时区，点击「创建」

验证与排查 #

• 手动上传测试CSV到Drive文件夹，在Cloud Functions控制台点击「测试函数」，验证数据是否成功追加到BigQuery
• 查看BigQuery的loaded_files_log表，确认已加载的文件记录
• 如果加载失败，查看Cloud Functions的日志（控制台→函数→日志），检查权限、CSV结构是否匹配等问题

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