Bruin 变量体系实战用内置日期变量与自定义变量参数化 Data Engineering 管道【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp本篇指南聚焦 Data Engineering Zoomcamp 第 5 模块「Data Platforms」中 Bruin 的核心概念之一——Variables变量。Bruin 会在每次管道运行时动态初始化变量让你用模板化与参数化的方式复用同一份管道代码内置的start_date/end_date随调度区间自动注入自定义变量则可在运行前按需覆盖。读完本文你将掌握在 SQL 资产中通过 Jinja 注入变量、在 Python 资产中通过环境变量读取变量、在pipeline.yml中声明自定义变量以及用--var、--start-date等 CLI 参数驱动运行时行为并能参照本仓库的纽约出租车NYC Taxi管道案例落地到自己的项目。变量是什么一次运行一份动态上下文在 Bruin 中Variables 是每次创建管道运行run时动态初始化的值。它们的存在是为了让管道可以被参数化同一份资产代码通过变量在不同时间窗口、不同客户、不同环境下产出不同结果而无需修改任何文件。从本仓库的课程笔记 06-core-04-variables.md 可以看到变量体系分为两大类内置变量Built-in Variables由 Bruin 在每次运行时自动提供无需声明自定义变量Custom Variables用户在pipeline.yml中定义可在创建运行时通过 CLI 覆盖默认值。这套机制与管道的三大核心概念项目、管道、资产紧密配合变量定义在管道层pipeline.yml消费在资产层SQL / Python 代码覆盖在运行层CLI 或 VS Code 扩展面板。理解这条链路是掌握 Bruin 参数化能力的关键。内置变量由调度区间决定的日期窗口Bruin 始终自动提供两个内置变量变量说明start_date调度区间的开始时间end_date调度区间的结束时间这两个日期的取值完全由管道的schedule决定笔记中给出了三种常见调度的映射关系调度开始日期结束日期Monthly每月当月第一天当月最后一天Daily每日当天开始时刻当天结束时刻Hourly每小时该小时开始时刻该小时结束时刻注意默认语义end_date默认是闭区间inclusive。如果你希望按半开区间处理例如按 start AND end过滤需要配合--exclusive-end-date使用这一点在后面的快速参考中会再次出现。SQL 资产中的注入Jinja 模板在 SQL 资产中变量通过Jinja 模板注入到查询语句里。这是最常见的用法——用日期窗口做增量抽取或分区裁剪bruin.asset(namestaging.monthly_trips, typesql) SELECT * FROM raw.trips WHERE pickup_date {{ start_date }} AND pickup_date {{ end_date }}书写时变量名不带前缀{{ start_date }}Bruin 在编译阶段会把它替换成真实值。你可以在 VS Code 的Bruin Render 面板中预览编译后的完整查询确认实际注入的值是否符合预期再决定是否运行。Python 资产中的访问环境变量在 Python 资产中内置变量以环境变量形式暴露命名规则为BRUIN_VAR_前缀 变量名大写import os from datetime import datetime bruin.asset(nameraw.monthly_data, typepython) def ingest_monthly_data(): start_date os.environ[BRUIN_VAR_START_DATE] end_date os.environ[BRUIN_VAR_END_DATE] # Parse and use dates to fetch data for specific period start datetime.fromisoformat(start_date) end datetime.fromisoformat(end_date) # Loop through months in range # ...拿到日期字符串后通过datetime.fromisoformat解析即可用于循环拉取指定时间段的 API 数据。仓库实战印证NYC Taxi 管道的日期窗口本仓库的端到端示例 03-nyc-taxi-pipeline.md 就是这一机制的完整落地。其 Python 摄取资产ingestion/trips.py这样读取运行区间import os import json import pandas as pd def materialize(): start_date os.environ[BRUIN_START_DATE] end_date os.environ[BRUIN_END_DATE] taxi_types json.loads(os.environ[BRUIN_VARS]).get(taxi_types, [yellow]) # ... return final_dataframe注意这里使用的是BRUIN_START_DATE/BRUIN_END_DATE不带VAR的命名变体随后脚本在起止日期之间按月生成日期列表去拉取https://d37ci6vzurychx.cloudfront.net/trip-data/{taxi_type}_tripdata_{year}-{month}.parquet格式的公开出租车数据。这印证了内置日期变量的核心应用把处理哪个时间段从代码里抽离出来交给运行参数决定。仓库中还说明了一个重要细节pipeline.yml里的start_date配置项如start_date: 2022-01-01决定的是全量刷新full refresh时从哪个日期开始处理数据与每次运行注入的区间变量是两套不同但协同的机制。自定义变量在 pipeline.yml 中声明自定义变量是用户定义的、作用域在管道级别的参数。它们声明在管道目录下的pipeline.yml中与调度、默认连接并列。声明格式variables: - name: taxi_types type: array default: - yellow声明包含三个要素变量名name、类型type如array、string和默认值default。仓库中的 NYC Taxi 示例 03-nyc-taxi-pipeline.md 给出了更完整的写法用items限定数组成员的类型variables: taxi_types: type: array items: type: string default: [yellow]这里taxi_types的作用是控制管道要摄取哪些出租车类型yellow / green默认可只取yellow运行时再决定是否扩展。运行时覆盖默认值创建运行时用--var覆盖自定义变量的默认值。语法为--var KEYVALUE值支持数组字面量bruin run ./pipeline.yml --var taxi_types[green,fhv]也可以覆盖字符串类型的变量bruin run ./pipeline.yml --var customer_id12345在 Python 中读取自定义变量与内置变量一样自定义变量在 Python 中也以BRUIN_VAR_前缀的环境变量形式暴露。由于值是 JSON 编码的数组类型需要用json.loads还原import os import json bruin.asset(nameexample.asset, typepython) def example_asset(): # Custom variables are prefixed with BRUIN_VAR_ taxi_types_json os.environ[BRUIN_VAR_TAXI_TYPES] taxi_types json.loads(taxi_types_json) # Use the variable in your code for taxi_type in taxi_types: # Process each taxi type pass仓库中的 NYC Taxi 示例采用了另一种等价做法读取BRUIN_VARS整体 JSON再用.get(taxi_types, [yellow])取出并附带默认值兜底。两种方式都可行BRUIN_VAR_前缀按单变量精确读取BRUIN_VARS则适合一次取多个变量并统一做默认值处理。变量与 VS Code 扩展面板Bruin 为 VS Code / Cursor 提供了扩展面板把变量操作集成进 IDE相关安装与 MCP 配置见 02-getting-started.md。面板主要提供三项能力Variable Override变量覆盖运行前直接设置自定义变量的值效果等同于 CLI 的--varBruin Render模板渲染预览实时查看 Jinja 模板被替换成真实值后的编译结果——这是排查日期窗口、引号、格式问题的最快途径Run Configuration运行配置集中设置日期、环境environment与变量一键发起运行。对于「参数化管道 模板调试」的工作流面板与 CLI 是互补的面板适合开发期的即时预览与试运行CLI 适合脚本化、定时化与 CI 中的正式执行。典型应用场景变量机制可以支撑以下几类常见的工程需求整理自原文档应用场景说明基于日期的分区处理为特定时间段抽取数据配合内置start_date/end_date精确裁剪时间窗口多租户处理同一管道为不同客户各跑一次通过--var customer_id...切换租户参数化转换根据变量改变转换逻辑例如按taxi_types决定处理哪些数据源A/B 测试不改代码仅通过变量切换不同配置组合进行对照实验快速参考常用运行命令以下是本仓库笔记中整理的常用命令速查表覆盖日期、变量与物化行为的控制# Run with custom dates自定义运行日期 bruin run ./pipeline.yml --start-date 2020-01-01 --end-date 2020-01-31 # Run with variable override (array)数组型变量覆盖 bruin run ./pipeline.yml --var taxi_types[green,fhv] # Run with variable override (string)字符串型变量覆盖 bruin run ./pipeline.yml --var customer_id12345 # Run with full refresh (affects materialization)全量刷新影响物化策略 bruin run ./pipeline.yml --full-refresh # Set end date as exclusiveend_date 按开区间处理 bruin run ./pipeline.yml --exclusive-end-date参数要点说明--start-date/--end-date覆盖本次运行的日期窗口直接影响注入到资产里的start_date/end_date--full-refresh会越过增量物化策略重建表与time_interval、append等策略的取舍可参见 06-core-03-assets.md 中的物化策略表--exclusive-end-date将end_date从默认的闭区间改为开区间配合WHERE col {{ start_date }} AND col {{ end_date }}这类半开区间过滤时语义更严谨。把变量放进完整的 Bruin 工作流变量不是孤立概念它串起了 Bruin 的核心执行链路各环节对应本模块的系列笔记项目Project → .bruin.yml环境、连接见 06-core-01-projects.md ↓ 管道Pipeline → pipeline.yml调度、默认连接、自定义变量见 06-core-02-pipelines.md ↓ 资产Assets → SQLJinja 注入/ Python环境变量读取/ Seed静态数据 ↓ 命令Commands → bruin run / validate / lineage / query见 06-core-05-commands.md一次典型运行中变量的完整生命周期是Bruin 依据调度区间初始化start_date/end_date→ 合并pipeline.yml中的自定义变量默认值 → 叠加 CLI 或面板传入的--var、--start-date等覆盖 → 按依赖顺序执行各资产并在编译 SQL、注入 Python 环境时把变量落到实处。理解这条链路后你就能把任何按时间窗口、按客户、按配置重复执行的管道改造成一套真正参数化、可复用的 Bruin 资产。进一步阅读仓库内配套资料06-core-04-variables.md本文原始课程笔记、03-nyc-taxi-pipeline.md变量实战的完整管道、06-core-05-commands.md运行选项详解。【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Bruin 变量体系实战：用内置日期变量与自定义变量参数化 Data Engineering 管道
Bruin 变量体系实战用内置日期变量与自定义变量参数化 Data Engineering 管道【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp本篇指南聚焦 Data Engineering Zoomcamp 第 5 模块「Data Platforms」中 Bruin 的核心概念之一——Variables变量。Bruin 会在每次管道运行时动态初始化变量让你用模板化与参数化的方式复用同一份管道代码内置的start_date/end_date随调度区间自动注入自定义变量则可在运行前按需覆盖。读完本文你将掌握在 SQL 资产中通过 Jinja 注入变量、在 Python 资产中通过环境变量读取变量、在pipeline.yml中声明自定义变量以及用--var、--start-date等 CLI 参数驱动运行时行为并能参照本仓库的纽约出租车NYC Taxi管道案例落地到自己的项目。变量是什么一次运行一份动态上下文在 Bruin 中Variables 是每次创建管道运行run时动态初始化的值。它们的存在是为了让管道可以被参数化同一份资产代码通过变量在不同时间窗口、不同客户、不同环境下产出不同结果而无需修改任何文件。从本仓库的课程笔记 06-core-04-variables.md 可以看到变量体系分为两大类内置变量Built-in Variables由 Bruin 在每次运行时自动提供无需声明自定义变量Custom Variables用户在pipeline.yml中定义可在创建运行时通过 CLI 覆盖默认值。这套机制与管道的三大核心概念项目、管道、资产紧密配合变量定义在管道层pipeline.yml消费在资产层SQL / Python 代码覆盖在运行层CLI 或 VS Code 扩展面板。理解这条链路是掌握 Bruin 参数化能力的关键。内置变量由调度区间决定的日期窗口Bruin 始终自动提供两个内置变量变量说明start_date调度区间的开始时间end_date调度区间的结束时间这两个日期的取值完全由管道的schedule决定笔记中给出了三种常见调度的映射关系调度开始日期结束日期Monthly每月当月第一天当月最后一天Daily每日当天开始时刻当天结束时刻Hourly每小时该小时开始时刻该小时结束时刻注意默认语义end_date默认是闭区间inclusive。如果你希望按半开区间处理例如按 start AND end过滤需要配合--exclusive-end-date使用这一点在后面的快速参考中会再次出现。SQL 资产中的注入Jinja 模板在 SQL 资产中变量通过Jinja 模板注入到查询语句里。这是最常见的用法——用日期窗口做增量抽取或分区裁剪bruin.asset(namestaging.monthly_trips, typesql) SELECT * FROM raw.trips WHERE pickup_date {{ start_date }} AND pickup_date {{ end_date }}书写时变量名不带前缀{{ start_date }}Bruin 在编译阶段会把它替换成真实值。你可以在 VS Code 的Bruin Render 面板中预览编译后的完整查询确认实际注入的值是否符合预期再决定是否运行。Python 资产中的访问环境变量在 Python 资产中内置变量以环境变量形式暴露命名规则为BRUIN_VAR_前缀 变量名大写import os from datetime import datetime bruin.asset(nameraw.monthly_data, typepython) def ingest_monthly_data(): start_date os.environ[BRUIN_VAR_START_DATE] end_date os.environ[BRUIN_VAR_END_DATE] # Parse and use dates to fetch data for specific period start datetime.fromisoformat(start_date) end datetime.fromisoformat(end_date) # Loop through months in range # ...拿到日期字符串后通过datetime.fromisoformat解析即可用于循环拉取指定时间段的 API 数据。仓库实战印证NYC Taxi 管道的日期窗口本仓库的端到端示例 03-nyc-taxi-pipeline.md 就是这一机制的完整落地。其 Python 摄取资产ingestion/trips.py这样读取运行区间import os import json import pandas as pd def materialize(): start_date os.environ[BRUIN_START_DATE] end_date os.environ[BRUIN_END_DATE] taxi_types json.loads(os.environ[BRUIN_VARS]).get(taxi_types, [yellow]) # ... return final_dataframe注意这里使用的是BRUIN_START_DATE/BRUIN_END_DATE不带VAR的命名变体随后脚本在起止日期之间按月生成日期列表去拉取https://d37ci6vzurychx.cloudfront.net/trip-data/{taxi_type}_tripdata_{year}-{month}.parquet格式的公开出租车数据。这印证了内置日期变量的核心应用把处理哪个时间段从代码里抽离出来交给运行参数决定。仓库中还说明了一个重要细节pipeline.yml里的start_date配置项如start_date: 2022-01-01决定的是全量刷新full refresh时从哪个日期开始处理数据与每次运行注入的区间变量是两套不同但协同的机制。自定义变量在 pipeline.yml 中声明自定义变量是用户定义的、作用域在管道级别的参数。它们声明在管道目录下的pipeline.yml中与调度、默认连接并列。声明格式variables: - name: taxi_types type: array default: - yellow声明包含三个要素变量名name、类型type如array、string和默认值default。仓库中的 NYC Taxi 示例 03-nyc-taxi-pipeline.md 给出了更完整的写法用items限定数组成员的类型variables: taxi_types: type: array items: type: string default: [yellow]这里taxi_types的作用是控制管道要摄取哪些出租车类型yellow / green默认可只取yellow运行时再决定是否扩展。运行时覆盖默认值创建运行时用--var覆盖自定义变量的默认值。语法为--var KEYVALUE值支持数组字面量bruin run ./pipeline.yml --var taxi_types[green,fhv]也可以覆盖字符串类型的变量bruin run ./pipeline.yml --var customer_id12345在 Python 中读取自定义变量与内置变量一样自定义变量在 Python 中也以BRUIN_VAR_前缀的环境变量形式暴露。由于值是 JSON 编码的数组类型需要用json.loads还原import os import json bruin.asset(nameexample.asset, typepython) def example_asset(): # Custom variables are prefixed with BRUIN_VAR_ taxi_types_json os.environ[BRUIN_VAR_TAXI_TYPES] taxi_types json.loads(taxi_types_json) # Use the variable in your code for taxi_type in taxi_types: # Process each taxi type pass仓库中的 NYC Taxi 示例采用了另一种等价做法读取BRUIN_VARS整体 JSON再用.get(taxi_types, [yellow])取出并附带默认值兜底。两种方式都可行BRUIN_VAR_前缀按单变量精确读取BRUIN_VARS则适合一次取多个变量并统一做默认值处理。变量与 VS Code 扩展面板Bruin 为 VS Code / Cursor 提供了扩展面板把变量操作集成进 IDE相关安装与 MCP 配置见 02-getting-started.md。面板主要提供三项能力Variable Override变量覆盖运行前直接设置自定义变量的值效果等同于 CLI 的--varBruin Render模板渲染预览实时查看 Jinja 模板被替换成真实值后的编译结果——这是排查日期窗口、引号、格式问题的最快途径Run Configuration运行配置集中设置日期、环境environment与变量一键发起运行。对于「参数化管道 模板调试」的工作流面板与 CLI 是互补的面板适合开发期的即时预览与试运行CLI 适合脚本化、定时化与 CI 中的正式执行。典型应用场景变量机制可以支撑以下几类常见的工程需求整理自原文档应用场景说明基于日期的分区处理为特定时间段抽取数据配合内置start_date/end_date精确裁剪时间窗口多租户处理同一管道为不同客户各跑一次通过--var customer_id...切换租户参数化转换根据变量改变转换逻辑例如按taxi_types决定处理哪些数据源A/B 测试不改代码仅通过变量切换不同配置组合进行对照实验快速参考常用运行命令以下是本仓库笔记中整理的常用命令速查表覆盖日期、变量与物化行为的控制# Run with custom dates自定义运行日期 bruin run ./pipeline.yml --start-date 2020-01-01 --end-date 2020-01-31 # Run with variable override (array)数组型变量覆盖 bruin run ./pipeline.yml --var taxi_types[green,fhv] # Run with variable override (string)字符串型变量覆盖 bruin run ./pipeline.yml --var customer_id12345 # Run with full refresh (affects materialization)全量刷新影响物化策略 bruin run ./pipeline.yml --full-refresh # Set end date as exclusiveend_date 按开区间处理 bruin run ./pipeline.yml --exclusive-end-date参数要点说明--start-date/--end-date覆盖本次运行的日期窗口直接影响注入到资产里的start_date/end_date--full-refresh会越过增量物化策略重建表与time_interval、append等策略的取舍可参见 06-core-03-assets.md 中的物化策略表--exclusive-end-date将end_date从默认的闭区间改为开区间配合WHERE col {{ start_date }} AND col {{ end_date }}这类半开区间过滤时语义更严谨。把变量放进完整的 Bruin 工作流变量不是孤立概念它串起了 Bruin 的核心执行链路各环节对应本模块的系列笔记项目Project → .bruin.yml环境、连接见 06-core-01-projects.md ↓ 管道Pipeline → pipeline.yml调度、默认连接、自定义变量见 06-core-02-pipelines.md ↓ 资产Assets → SQLJinja 注入/ Python环境变量读取/ Seed静态数据 ↓ 命令Commands → bruin run / validate / lineage / query见 06-core-05-commands.md一次典型运行中变量的完整生命周期是Bruin 依据调度区间初始化start_date/end_date→ 合并pipeline.yml中的自定义变量默认值 → 叠加 CLI 或面板传入的--var、--start-date等覆盖 → 按依赖顺序执行各资产并在编译 SQL、注入 Python 环境时把变量落到实处。理解这条链路后你就能把任何按时间窗口、按客户、按配置重复执行的管道改造成一套真正参数化、可复用的 Bruin 资产。进一步阅读仓库内配套资料06-core-04-variables.md本文原始课程笔记、03-nyc-taxi-pipeline.md变量实战的完整管道、06-core-05-commands.md运行选项详解。【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考