Snowflake 数据仓库中的模式版本控制

Posted 2023-02-16

【中文标题】Snowflake 数据仓库中的模式版本控制

【英文标题】：Schema versioning within the Snowflake data warehouse

【发布时间】：2020-04-03 22:14:42

【问题描述】：

我对通过使用模式版本控制使雪花数据库的用户免受变化影响的方式很感兴趣。我一直在研究使用连接语法来定义一个模式，其中将为每个版本创建一个包含核心表视图的新模式，任何未更改的视图都将被复制其他经过修改的视图将向后兼容。理想情况下，用户在连接时会获得所需版本的正确连接语法。

我遇到的问题是，有多个团队各自拥有与核心业务领域相关的模式，我认为不可能在连接语法中定义多个模式。

是否有人在具有多个用户、架构和开发团队的环境中实现了这一目标？

问候，

卢克

【问题讨论】：

我听说过许多外部工具（GitHub、CircleCI 等），客户在他们的 Snowflake 工作流程中实施以更改对模式的控制。我不认为这些工具现在直接集成到 Snowflake 中，但这将是一个很好的功能要求。

是的，我确实遵循相同的过程。每次我在雪花中创建或修改表、视图、存储过程时，我都会更改并提交到 Git 存储库。所以我已经完成了更改的历史数据。当您拥有庞大的团队并从事相同的流程时，此流程也变得很重要。但就像@SuzyLockwood 所说，没有直接集成到雪花中的工具，我们只需要实现自己的处理流程即可。

【参考方案1】：

我们使用 Alembic 来控制 Snowflake 的数据库版本。 Alembic 是一个“迁移”工具，您可以在其中对数据仓库运行多项更改（或迁移）。它本质上是 Python 中 SQLAlchemy 库的附加组件。

在本地开发时，我们创建数据库的克隆，并测试我们对克隆数据库的迁移更改。一旦我们知道它有效，我们将其推送到 GitLab，获得批准，然后我们可以运行具有 accountadmin 凭据的 CI/CD 管道，以在生产环境中进行更改。

由于它是用 Python 编写的，因此您可以将其连接到您的 Git 工具（如 GitHub 或 GitLab）并在合并请求中提交更改并在您的生产数据库中运行它之前获得批准。

这是文档：https://alembic.sqlalchemy.org/en/latest/

根据 Snowflake 文档，这也得到官方支持：https://docs.snowflake.net/manuals/user-guide/sqlalchemy.html#alembic-support

Alembic 迁移示例可能如下所示：

Revision ID: 78a3acc7fbb2
Revises: 3f2ee8d809a6
Create Date: 2019-11-06 11:40:38.438468

"""

# revision identifiers, used by Alembic.
revision = '78a3acc7fbb2'
down_revision = '3f2ee8d809a6'
branch_labels = None
depends_on = None

from alembic import op
import sqlalchemy as sa

def upgrade():
    op.create_table('test_table',
    sa.Column('op', sa.String(length=255), nullable=True),
    sa.Column('id', sa.String(length=255), nullable=False),
    sa.Column('amount', sa.BigInteger(), nullable=True),
    sa.Column('reason', sa.String(length=255), nullable=True),
    sa.Column('deleted', sa.Boolean(), nullable=True),
    sa.Column('user_id', sa.Integer(), nullable=True),
    sa.Column('company_id', sa.Integer(), nullable=True),
    sa.Column('inserted_at', sa.DateTime(), nullable=True),
    sa.Column('updated_at', sa.DateTime(), nullable=True),
    sa.Column('dw_import_filename', sa.String(length=255), nullable=True),
    sa.Column('dw_import_file_row_number', sa.Integer(), nullable=True),
    sa.Column('dw_import_timestamp', sa.TIMESTAMP(), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    schema='test_schema'
    )

def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('test_table', schema='test_schema')

如您所见，您必须提供升级并具有降级的能力，这与升级相反。如果您对 Alembic 有任何其他问题，或者您对此感兴趣，那么我很乐意解释更多。

【参考方案2】：

作为发布过程的第一步，您可以将“当前版本”架构克隆到新的“x 版本”架构，然后使用原始名称部署新架构。

示例：

CREATE SCHEMA MY_DB.MY_SCHEMA_20200103 CLONE MY_DB.MY_SCHEMA;

...部署发布步骤...

重要提示

您必须小心所有引用，以确保它们指向正确的对象，无论是版本化架构中的对象还是当前架构中的对象。 这将复制架构内所有对象的授权，但需要授予新创建架构的权限