如何修改BigQuery动态Upsert方案，用EXTERNAL_QUERY对接Cloud SQL

一、核心改造思路 #

原方案的核心是通过EXECUTE IMMEDIATE动态生成UPSERT语句，只需将数据源替换为EXTERNAL_QUERY()查询Cloud SQL的增量变更数据，结合同步时间控制逻辑，即可实现定期将Cloud SQL的新增/更新数据同步到BigQuery。

二、具体改造步骤 #

1. 定义增量同步规则 #

首先确定Cloud SQL表中用于识别变更的字段（比如updated_at时间戳、id自增主键），每次仅拉取上次同步后的数据，避免全量同步浪费资源。建议在BigQuery中创建一张同步控制表（如sync_control），用于存储上次同步的时间戳。

2. 用EXTERNAL_QUERY()获取Cloud SQL变更数据 #

将原方案中的数据源替换为EXTERNAL_QUERY()，调用已建立的Cloud SQL外部连接，拉取增量数据：

DECLARE cloud_sql_conn STRING;
DECLARE last_sync_ts TIMESTAMP;
DECLARE cloud_sql_query STRING;

-- 配置Cloud SQL外部连接名
SET cloud_sql_conn = 'your-cloud-sql-connection-name';

-- 从控制表获取上次同步时间（首次同步可设为'1970-01-01'）
SET last_sync_ts = (SELECT IFNULL(MAX(last_sync_time), TIMESTAMP('1970-01-01')) FROM `your-project.your-dataset.sync_control`);

-- 构造Cloud SQL增量查询语句，仅拉取变更数据
SET cloud_sql_query = '''
    SELECT id, col1, col2, updated_at 
    FROM your_cloud_sql_table 
    WHERE updated_at > ''''' || FORMAT_TIMESTAMP('%Y-%m-%d %H:%M:%S', last_sync_ts) || '''''';

-- 抽取Cloud SQL变更数据
WITH change_data AS (
    SELECT * FROM EXTERNAL_QUERY(cloud_sql_conn, cloud_sql_query)
)

3. 适配动态UPSERT逻辑 #

将change_data作为数据源，嵌入原方案的动态MERGE语句中，动态生成字段映射和更新逻辑，避免硬编码字段：

-- 定义BigQuery目标表
DECLARE target_table STRING;
SET target_table = 'your-project.your-dataset.target_bq_table';

-- 动态获取目标表的非主键字段（假设id是主键）
DECLARE non_pk_fields STRING;
SET non_pk_fields = (
    SELECT STRING_AGG(column_name, ', ')
    FROM `your-project.your-dataset.INFORMATION_SCHEMA.COLUMNS`
    WHERE table_name = 'target_bq_table'
      AND column_name != 'id'
);

-- 动态生成UPSERT语句
DECLARE upsert_sql STRING;
SET upsert_sql = '''
    MERGE `''' || target_table || '''` AS target
    USING change_data AS source
    ON target.id = source.id
    WHEN MATCHED THEN
        UPDATE SET ''' || non_pk_fields || ''' = source.''' || REPLACE(non_pk_fields, ', ', ', source.') || '''
    WHEN NOT MATCHED THEN
        INSERT (id, ''' || non_pk_fields || ''')
        VALUES (source.id, source.''' || REPLACE(non_pk_fields, ', ', ', source.') || ''')
''';

-- 执行动态UPSERT
EXECUTE IMMEDIATE upsert_sql;

-- 更新同步控制表的时间戳
MERGE `your-project.your-dataset.sync_control` AS sc
USING (SELECT CURRENT_TIMESTAMP() AS new_sync_time) AS s
ON 1=1 -- 控制表仅存一条同步记录
WHEN MATCHED THEN
    UPDATE SET last_sync_time = s.new_sync_time
WHEN NOT MATCHED THEN
    INSERT (last_sync_time)
    VALUES (s.new_sync_time);

4. 配置定期执行 #

将上述完整SQL脚本保存为BigQuery查询，通过BigQuery调度查询设置执行频率（如每小时、每天），实现自动定期同步。

三、关键注意事项 #

• 字段类型兼容：确保Cloud SQL返回的字段类型与BigQuery目标表字段类型一致，必要时添加类型转换逻辑。
• 主键唯一性：明确主键字段（如id），保证MERGE的ON条件能精准匹配需更新/插入的记录。
• 性能优化：给Cloud SQL中用于增量筛选的字段（如updated_at）添加索引，避免全表扫描。
• 权限配置：确保执行查询的账号拥有Cloud SQL外部连接权限，以及BigQuery目标表的读写权限。

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