每日将BigQuery同结构新表数据合并至主表的实现方法

假设你的新表命名格式为data_YYYYMMDD（比如data_20240520），和主表main_table同属一个数据集且结构完全一致，以下是两种可靠的实现方案：

方案一：BigQuery预定查询（适合简单场景） #

如果不需要复杂的前置校验（比如检查表是否存在），直接用BigQuery内置的预定查询就能快速搞定：

1. 编写动态合并SQL
   若只需追加数据（确认无重复），可以用动态表名生成逻辑自动匹配前一天的新表：

   DECLARE new_table_name STRING;
   -- 生成前一天的新表名称（根据实际命名格式调整）
   SET new_table_name = FORMAT("`your-project.your-dataset.data_%s`", FORMAT_DATE("%Y%m%d", DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)));
   
   -- 执行追加插入
   EXECUTE IMMEDIATE FORMAT("""
   INSERT INTO `your-project.your-dataset.main_table`
   SELECT * FROM %s
   """, new_table_name);
   

   如果需要避免重复数据（比如有主键），改用MERGE语句：

   DECLARE new_table_name STRING;
   SET new_table_name = FORMAT("`your-project.your-dataset.data_%s`", FORMAT_DATE("%Y%m%d", DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)));
   
   EXECUTE IMMEDIATE FORMAT("""
   MERGE `your-project.your-dataset.main_table` AS target
   USING %s AS source
   ON target.your_primary_key = source.your_primary_key  -- 替换成你的主键字段
   WHEN NOT MATCHED THEN
     INSERT ROW
   """, new_table_name);
   

2. 设置预定查询
   在BigQuery控制台中创建预定查询，设置每日执行时间（确保在新表上传完成之后），绑定上述SQL即可自动执行。

方案二：Cloud Functions + Cloud Scheduler（适合需要校验的场景） #

如果需要先确认新表是否存在再执行合并，或者有其他自定义逻辑，用Cloud Functions结合Cloud Scheduler更灵活：

1. 编写Python云函数

   from google.cloud import bigquery
   import datetime
   
   def merge_new_table(event, context):
       client = bigquery.Client()
       project_id = "your-project"
       dataset_id = "your-dataset"
       main_table_full_id = f"{project_id}.{dataset_id}.main_table"
       
       # 生成前一天的新表ID
       yesterday = datetime.date.today() - datetime.timedelta(days=1)
       new_table_suffix = yesterday.strftime("%Y%m%d")
       new_table_full_id = f"{project_id}.{dataset_id}.data_{new_table_suffix}"
       
       # 校验新表是否存在
       try:
           client.get_table(new_table_full_id)
       except Exception as e:
           print(f"跳过合并：新表 {new_table_full_id} 不存在，错误信息: {str(e)}")
           return
       
       # 执行合并（需要去重的话替换成MERGE逻辑）
       merge_query = f"""
       INSERT INTO `{main_table_full_id}`
       SELECT * FROM `{new_table_full_id}`
       """
       query_job = client.query(merge_query)
       query_job.result()  # 等待查询执行完成
       print(f"合并完成：已将 {new_table_full_id} 的数据写入主表")
   

2. 部署与调度

   • 将上述代码部署为Cloud Functions，确保服务账号拥有BigQuery的读写权限
   • 在Cloud Scheduler中创建每日触发任务，调用该云函数的HTTP触发器

注意事项 #

• 权限配置：确保执行任务的账号（预定查询/云函数的服务账号）拥有新表的读取权限和主表的写入权限
• 时间调整：如果新表是当日上传的，把DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)改成CURRENT_DATE()即可，根据实际数据上传时间调整日期逻辑
• 数据一致性：如果主表和新表有字段更新需求，建议明确指定字段列表，避免结构变更导致的问题

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