如何在分支部署中克隆生产 Snowflake 数据测试代码而不影响生产？

如何在分支部署中克隆生产 Snowflake 数据测试代码而不影响生产【免费下载链接】dagsterAn orchestration platform for the development, production, and observation of data assets.项目地址: https://gitcode.com/GitHub_Trending/da/dagster在 Dagster 中功能分支的代码往往需要在接近生产的数据上跑通才能合入但直接读写生产库风险很高。本文基于 Dagster 仓库中的官方指南testing-against-prod-data.md演示一套完整可执行的做法利用分支部署branch deployments Snowflake 数据库克隆让每次拉取请求PR自动获得一个PRODUCTION_CLONE_ID数据库克隆ID为 PR ID分支部署中的资产只读写该克隆PR 关闭后自动删除克隆。整个过程使用与生产不同的连接配置和数据库名确保分支部署不会写到PRODUCTION库。准备条件按文档要求你需要一个 Dagster 账号一个用create-dagster projectCLI 创建的空 Dagster 项目参见 Creating Dagster projects项目中已配置好的分支部署使用 GitHub Actions.github/workflows/dagster-plus-deploy.ymlServerless或dagster-cloud-deploy.ymlHybrid/ GitLab CI.gitlab-ci.ymlHybrid 部署还需要一个安装项目的 Dockerfile你在 Dagster 中具有访问分支部署的权限。文档中的示例场景生产端有一个PRODUCTIONSnowflake 数据库schema 为HACKER_NEWS其中包含ITEMS、COMMENTS、STORIES三张 Hacker News 相关表。第 1 步把 Snowflake 表定义成资产三个资产都通过io_manager_keysnowflake_io_manager声明持久化目标。这样资产本身不关心写到哪里写入哪个数据库完全由 I/O manager 的配置决定——这是后面按环境切换连接的关键。src/project_name/defs/assets.py# assets.py import pandas as pd import requests import dagster as dg class ItemsConfig(dg.Config): base_item_id: int dg.asset( io_manager_keysnowflake_io_manager, ) def items(config: ItemsConfig) - pd.DataFrame: Items from the Hacker News API: each is a story or a comment on a story. rows [] max_id requests.get( https://hacker-news.firebaseio.com/v0/maxitem.json, timeout5 ).json() # Hacker News API is 1-indexed, so adjust range by 1 for item_id in range(max_id - config.base_item_id 1, max_id 1): item_url fhttps://hacker-news.firebaseio.com/v0/item/{item_id}.json rows.append(requests.get(item_url, timeout5).json()) # ITEM_FIELD_NAMES is a list of the column names in the Hacker News dataset result pd.DataFrame(rows, columnsITEM_FIELD_NAMES).drop_duplicates(subset[id]) result.rename(columns{by: user_id}, inplaceTrue) return result dg.asset( io_manager_keysnowflake_io_manager, ) def comments(items: pd.DataFrame) - pd.DataFrame: Comments from the Hacker News API. return items[items[type] comment] dg.asset( io_manager_keysnowflake_io_manager, ) def stories(items: pd.DataFrame) - pd.DataFrame: Stories from the Hacker News API. return items[items[type] story]其中ITEM_FIELD_NAMES是列名列表id,parent,time,type,by,text,kids,score,title,descendants,url完整定义见仓库中的 assets.py。Snowflake I/O manager 的用法可参考 using-snowflake-with-dagster-io-managers。第 2 步按环境配置 I/O manager 与资源运行时通过读取 Dagster 自动设置的内置环境变量判断当前环境DAGSTER_CLOUD_IS_BRANCH_DEPLOYMENT值为1时表示当前是分支部署DAGSTER_CLOUD_PULL_REQUEST_ID分支部署中可用的 PR ID。为防止分支部署误写生产文档的做法是分支环境使用与生产不同的凭证并写入以 PR ID 命名的克隆库生产环境写入PRODUCTION。最终的资源配置如下文档分两步给出第 2 步先配 I/O manager第 3 步加入执行 SQL 的SnowflakeResource这里是合并后的最终版本src/project_name/defs/resources.pyimport os from dagster_snowflake import SnowflakeResource from dagster_snowflake_pandas import SnowflakePandasIOManager import dagster as dg from .clone_and_drop_db import clone_prod, drop_prod_clone snowflake_config { account: {env: SNOWFLAKE_ACCOUNT}, user: {env: SNOWFLAKE_USER}, password: {env: SNOWFLAKE_PASSWORD}, schema: HACKER_NEWS, } resources { branch: { snowflake_io_manager: SnowflakePandasIOManager( **snowflake_config, databasefPRODUCTION_CLONE_{os.getenv(DAGSTER_CLOUD_PULL_REQUEST_ID)}, ), snowflake: SnowflakeResource( **snowflake_config, databasefPRODUCTION_CLONE_{os.getenv(DAGSTER_CLOUD_PULL_REQUEST_ID)}, ), }, prod: { snowflake_io_manager: SnowflakePandasIOManager( **snowflake_config, databasePRODUCTION, ), snowflake: SnowflakeResource(**snowflake_config, databasePRODUCTION), }, } def get_current_env(): is_branch_depl os.getenv(DAGSTER_CLOUD_IS_BRANCH_DEPLOYMENT) assert is_branch_depl is not None # env var must be set return branch if is_branch_depl else prod dg.definitions def resources(): return dg.Definitions( resourcesresources[get_current_env()], )注意区分两个 Snowflake 组件的职责SnowflakePandasIOManager负责把资产输出写入 SnowflakeSnowflakeResource负责执行查询克隆/删除数据库的 op 会用到它。两者使用相同的连接配置。SNOWFLAKE_ACCOUNT/SNOWFLAKE_USER/SNOWFLAKE_PASSWORD是文档中通过{env: ...}引用的环境变量名你需要在 Dagster 中配置对应变量生产与分支建议使用不同凭证。文档早期版本还演示过把account、user直接写死、仅password走环境变量的写法两种都来自源文档按需选择。第 3 步编写克隆与删除数据库的作业文档特意用 op/job 而非资产来实现克隆和删除理由是两个任务只关心克隆/删除数据库这个动作本身不需要资产图、回填等资产特性。src/project_name/defs/clone_and_drop_db.pyimport os from dagster_snowflake import SnowflakeResource import dagster as dg dg.op def drop_database_clone(snowflake: SnowflakeResource): with snowflake.get_connection() as conn: cur conn.cursor() cur.execute( DROP DATABASE IF EXISTS f PRODUCTION_CLONE_{os.environ[DAGSTER_CLOUD_PULL_REQUEST_ID]} ) dg.op(ins{start: dg.In(dg.Nothing)}) def clone_production_database(snowflake: SnowflakeResource): with snowflake.get_connection() as conn: cur conn.cursor() cur.execute( CREATE DATABASE f PRODUCTION_CLONE_{os.environ[DAGSTER_CLOUD_PULL_REQUEST_ID]} CLONE PRODUCTION ) dg.graph def clone_prod(): clone_production_database(startdrop_database_clone()) dg.graph def drop_prod_clone(): drop_database_clone()clone_prod的逻辑是先DROP DATABASE IF EXISTS再CREATE ... CLONE保证每次重新部署向分支再次 push都拿到一份全新的PRODUCTION克隆drop_prod_clone用于分支合并后清理。再把这两个 graph 绑定为 job并且只在分支部署环境中注册避免生产部署里出现这两个作业src/project_name/defs/jobs.pybranch_deployment_jobs [ clone_prod.to_job(), drop_prod_clone.to_job(), ] dg.definitions def jobs(): return dg.Definitions( jobs( branch_deployment_jobs if os.getenv(DAGSTER_CLOUD_IS_BRANCH_DEPLOYMENT) 1 else [] ), )第 4 步在 CI 中于 PR 打开时触发克隆使用 GitHub Actions在已有的.github/workflows/dagster-plus-deploy.ymlServerless或dagster-cloud-deploy.ymlHybrid的dagster_cloud_build_push作业末尾追加一步在部署启动后向该分支部署排队一次clone_prod运行。下面的代码块保留了源文档中的省略号它代表你现有 workflow 中已有的步骤只把最后一个 step 追加进去即可if条件保证 PR 关闭时不会触发克隆name: Dagster Branch Deployments on: pull_request: types: [opened, synchronize, reopened, closed] env: DAGSTER_CLOUD_URL: ${{ secrets.DAGSTER_CLOUD_URL }} jobs: dagster_cloud_build_push: runs-on: ubuntu-latest name: Dagster Branch Deployments strategy: ... steps: # Existing steps here ... - name: Clone Snowflake schema upon launch if: github.event.action ! closed uses: dagster-io/dagster-cloud-action/actions/utils/runv0.1 with: location_name: ${{ matrix.location.name }} deployment: ${{ steps.deploy.outputs.deployment }} job_name: clone_prod env: DAGSTER_CLOUD_URL: ${{ secrets.DAGSTER_CLOUD_URL }} DAGSTER_CLOUD_API_TOKEN: ${{ secrets.DAGSTER_CLOUD_API_TOKEN }}该作业由opened、synchronize、reopened、closed四类 PR 事件触发后续向分支 push 新提交会再次触发clone_prod得到一份新的克隆。使用 GitLab CI/CD可选分支如果项目使用 GitLab则在.gitlab-ci.yml的deploy作业中追加一个dagster-plus job launch步骤。该作业在 merge request 创建或更新时触发下面代码块中的省略号同样代表已有步骤--location参数需要替换为你存放clone_prod作业的 location 名deploy-docker-branch: stage: deploy rules: - if: $CI_PIPELINE_SOURCE merge_request_event dependencies: - build-image - parse-workspace image: ghcr.io/dagster-io/dagster-cloud-action:0.1.23 script: # Existing steps here ... # Add a step to launch the job cloning the prod db - dagster-plus job launch --url $DAGSTER_CLOUD_URL/$DEPLOYMENT_NAME --api-token $DAGSTER_CLOUD_API_TOKEN --location location_name_containing_clone_prod_job --job clone_prod environment: name: branch/$CI_COMMIT_REF_NAME on_stop: close_branch验证克隆已创建、资产写入克隆库打开 PR 后分支部署自动启动然后在 Dagster 中确认clone_prod作业已经运行也可以到 GitHub PR 的Actions标签页查看分支部署 workflow 日志在 Snowflake 中确认PRODUCTION_CLONE_PR ID数据库存在即每个分支部署都有一份对应克隆在分支部署中 materialize 资产后由于 I/O manager 指向克隆库资产输出会写入PRODUCTION_CLONE_ID。对克隆库执行查询即可验证分支代码产出的数据是否正确。第 5 步PR 关闭时删除克隆使用 GitHub Actions在同一个 workflow 文件中再追加一个 step仅在closed事件时排队drop_prod_clone运行省略号代表已有的步骤与克隆 stepname: Dagster Branch Deployments on: pull_request: types: [opened, synchronize, reopened, closed] env: DAGSTER_CLOUD_URL: ${{ secrets.DAGSTER_CLOUD_URL }} jobs: dagster_cloud_build_push: runs-on: ubuntu-latest name: Dagster Branch Deployments strategy: ... steps: # Existing steps here ... - name: Clone Snowflake schema upon launch ... - name: Delete schema clone upon PR close if: github.event.action closed uses: dagster-io/dagster-cloud-action/actions/utils/runv0.1 with: location_name: ${{ matrix.location.name }} deployment: ${{ steps.deploy.outputs.deployment }} job_name: drop_prod_clone env: DAGSTER_CLOUD_URL: ${{ secrets.DAGSTER_CLOUD_URL }} DAGSTER_CLOUD_API_TOKEN: ${{ secrets.DAGSTER_CLOUD_API_TOKEN }}使用 GitLab CI/CD可选分支在.gitlab-ci.yml中追加一个手动触发的close_branch作业用于排队drop_prod_clone运行--location替换为存放该作业的 location 名close_branch: stage: deploy image: ghcr.io/dagster-io/dagster-cloud-action:0.1.23 when: manual only: - merge_requests script: # Existing steps here ... # Add a step to launch the job dropping the cloned db - dagster-plus job launch --url $DAGSTER_CLOUD_URL/$DEPLOYMENT_NAME --api-token $DAGSTER_CLOUD_API_TOKEN --location location_name_containing_drop_prod_clone_job --job drop_prod_clone environment: name: branch/$CI_COMMIT_REF_NAME action: stop合并分支后到 Snowflake 中查看数据库列表确认对应克隆已被删除即清理完成。限制与说明该方案依赖 Dagster 分支部署能力文档标注为 Dagster 功能以及 CI 中dagster-io/dagster-cloud-action的run子动作 /dagster-plus job launch命令向分支部署排队作业隔离生产靠两层机制分支环境使用独立凭证且database参数固定为PRODUCTION_CLONE_PR ID两者任一失效都可能让写入落到其他库配置后应先在 Snowflake 中确认克隆库名DAGSTER_CLOUD_IS_BRANCH_DEPLOYMENT在所有部署中都会被设置而DAGSTER_CLOUD_PULL_REQUEST_ID仅在分支部署中可用相关取值见内置环境变量文档仓库中完整的示例代码可对照 dev_to_prod 目录 查看。【免费下载链接】dagsterAn orchestration platform for the development, production, and observation of data assets.项目地址: https://gitcode.com/GitHub_Trending/da/dagster创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考