基于GitLab-CI部署Snowflake数据库的版本追踪与回滚方案咨询

针对多Snowflake账户的版本追踪和回滚需求，推荐采用版本化迁移脚本+GitLab-CI自动化流水线+Snowflake元数据追踪的方案，以下是具体实现思路和示例：

一、核心规范：版本化迁移脚本结构 #

在GitLab仓库中统一管理迁移和回滚脚本，按版本号命名确保执行顺序可控：

snowflake-deploy/
├── migrations/          # 正向迁移脚本
│   ├── v1.0.0__init_core_schema.sql
│   ├── v1.0.1__create_orders_table.sql
│   └── v1.0.2__add_order_status_column.sql
├── rollbacks/           # 对应版本的回滚脚本
│   ├── rollback_v1.0.1__drop_orders_table.sql
│   └── rollback_v1.0.2__drop_order_status_column.sql
├── deploy.py            # 部署执行脚本
└── rollback.py          # 回滚执行脚本

• 脚本命名规则：v[版本号]__[描述].sql，版本号采用语义化格式（如v1.0.0）
• 每个迁移脚本必须对应一个回滚脚本，确保结构/数据变更可逆向操作

二、版本追踪实现：schema_migrations元数据表 #

在每个Snowflake账户的目标数据库中创建元数据表，记录已执行的迁移版本、流水线信息：

CREATE TABLE IF NOT EXISTS schema_migrations (
    version VARCHAR(255) PRIMARY KEY,
    executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    ci_pipeline_id VARCHAR(255),
    commit_hash VARCHAR(255)
);

部署脚本会读取该表，跳过已执行的迁移，确保幂等性；同时关联Git提交哈希和CI流水线ID，实现代码版本与数据库版本的双向追踪。

三、GitLab-CI流水线配置示例 #

.gitlab-ci.yml 核心配置 #

stages:
  - test
  - deploy
  - rollback

# 测试阶段：校验SQL脚本语法（用测试环境Snowflake账户）
test-migrations:
  stage: test
  image: python:3.10
  before_script:
    - pip install snowflake-connector-python
  script:
    - python -c "
import snowflake.connector, os, glob
conn = snowflake.connector.connect(
    user=os.environ['SNOWFLAKE_TEST_USER'],
    password=os.environ['SNOWFLAKE_TEST_PASSWORD'],
    account=os.environ['SNOWFLAKE_TEST_ACCOUNT'],
    warehouse=os.environ['SNOWFLAKE_TEST_WH'],
    database=os.environ['SNOWFLAKE_TEST_DB'],
    schema=os.environ['SNOWFLAKE_TEST_SCHEMA']
)
for file in glob('migrations/*.sql') + glob('rollbacks/*.sql'):
    with open(file, 'r') as f:
        conn.cursor().execute('EXPLAIN ' + f.read())
conn.close()
"
  only:
    - branches
    - tags

# 部署阶段：仅Git标签触发，部署到生产Snowflake账户
deploy-prod:
  stage: deploy
  image: python:3.10
  before_script:
    - pip install snowflake-connector-python
  script:
    - python deploy.py
  only:
    - tags
  variables:
    SNOWFLAKE_USER: $SNOWFLAKE_PROD_USER
    SNOWFLAKE_PASSWORD: $SNOWFLAKE_PROD_PASSWORD
    SNOWFLAKE_ACCOUNT: $SNOWFLAKE_PROD_ACCOUNT
    # 其他生产环境变量（仓库、数据库、Schema）

# 回滚阶段：手动触发，指定目标版本
rollback-prod:
  stage: rollback
  image: python:3.10
  before_script:
    - pip install snowflake-connector-python
  script:
    - python rollback.py --target-version $TARGET_VERSION
  when: manual
  variables:
    SNOWFLAKE_USER: $SNOWFLAKE_PROD_USER
    SNOWFLAKE_PASSWORD: $SNOWFLAKE_PROD_PASSWORD
    SNOWFLAKE_ACCOUNT: $SNOWFLAKE_PROD_ACCOUNT

• 多账户管理：通过GitLab CI/CD变量分组存储不同账户的连接信息（如SNOWFLAKE_PROD_*、SNOWFLAKE_DEV_*），不同job对应不同环境
• 安全控制：使用GitLab保护变量、保护分支，仅授权用户可触发部署/回滚流水线

四、核心脚本示例 #

deploy.py 部署脚本核心逻辑 #

import snowflake.connector
import os
from glob import glob

# 从CI变量获取Snowflake连接信息
conn = snowflake.connector.connect(
    user=os.environ['SNOWFLAKE_USER'],
    password=os.environ['SNOWFLAKE_PASSWORD'],
    account=os.environ['SNOWFLAKE_ACCOUNT'],
    warehouse=os.environ['SNOWFLAKE_WAREHOUSE'],
    database=os.environ['SNOWFLAKE_DATABASE'],
    schema=os.environ['SNOWFLAKE_SCHEMA']
)

# 初始化元数据表
with conn.cursor() as cur:
    cur.execute("""
        CREATE TABLE IF NOT EXISTS schema_migrations (
            version VARCHAR(255) PRIMARY KEY,
            executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            ci_pipeline_id VARCHAR(255),
            commit_hash VARCHAR(255)
        )
    """)

# 获取已执行版本
with conn.cursor() as cur:
    cur.execute("SELECT version FROM schema_migrations ORDER BY version")
    executed_versions = [row[0] for row in cur.fetchall()]

# 按版本顺序执行未执行的迁移
migration_files = sorted(glob('migrations/v*__*.sql'))
for file in migration_files:
    version = os.path.basename(file).split('__')[0]
    if version not in executed_versions:
        print(f"Executing migration: {version}")
        with open(file, 'r') as f:
            conn.cursor().execute(f.read())
        # 记录版本到元数据表
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO schema_migrations (version, ci_pipeline_id, commit_hash)
                VALUES (%s, %s, %s)
            """, (version, os.environ['CI_PIPELINE_ID'], os.environ['CI_COMMIT_SHA']))

conn.close()

rollback.py 回滚脚本核心逻辑 #

import snowflake.connector
import os
import argparse
from glob import glob

parser = argparse.ArgumentParser()
parser.add_argument('--target-version', required=True, help='Version to roll back to')
args = parser.parse_args()

conn = snowflake.connector.connect(
    user=os.environ['SNOWFLAKE_USER'],
    password=os.environ['SNOWFLAKE_PASSWORD'],
    account=os.environ['SNOWFLAKE_ACCOUNT'],
    warehouse=os.environ['SNOWFLAKE_WAREHOUSE'],
    database=os.environ['SNOWFLAKE_DATABASE'],
    schema=os.environ['SNOWFLAKE_SCHEMA']
)

# 获取已执行版本（逆序）
with conn.cursor() as cur:
    cur.execute("SELECT version FROM schema_migrations ORDER BY version DESC")
    executed_versions = [row[0] for row in cur.fetchall()]

# 筛选需要回滚的版本（所有高于目标版本的记录）
target_version = args.target_version
versions_to_rollback = [v for v in executed_versions if v > target_version]

# 按逆序执行回滚脚本
for version in versions_to_rollback:
    rollback_file = glob(f'rollbacks/rollback_{version}__*.sql')[0]
    print(f"Rolling back version: {version}")
    with open(rollback_file, 'r') as f:
        conn.cursor().execute(f.read())
    # 从元数据表删除该版本记录
    with conn.cursor() as cur:
        cur.execute("DELETE FROM schema_migrations WHERE version = %s", (version,))

conn.close()

五、存储过程的特殊处理 #

• 存储过程的创建/修改需纳入版本化管理，迁移脚本中完整定义存储过程代码
• 回滚脚本需恢复到上一版本的存储过程（可将历史版本代码备份在迁移脚本中），例如：

  -- 回滚脚本：恢复v1.0.2版本的存储过程
  DROP PROCEDURE IF EXISTS calculate_order_total();
  CREATE PROCEDURE calculate_order_total(ORDER_ID INT)
  RETURNS FLOAT
  LANGUAGE JAVASCRIPT
  AS $$
  -- 此处为v1.0.1版本的存储过程代码
  $$;
  

六、数据回滚补充方案 #

若需恢复误操作的数据，可结合Snowflake的时间旅行特性，在回滚脚本中添加数据恢复逻辑：

-- 恢复1小时前的orders表数据
INSERT INTO orders 
SELECT * FROM orders AT(OFFSET => -3600)
WHERE order_id NOT IN (SELECT order_id FROM orders);

注意：需确保Snowflake账户的数据保留期足够（默认1天，可配置最长90天）

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