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 始终自动提供两个内置变量：

这两个日期的取值完全由管道的schedule决定，笔记中给出了三种常见调度的映射关系：

注意默认语义：end_date默认是闭区间（inclusive）。如果你希望按半开区间处理（例如按>= start AND < end过滤），需要配合--exclusive-end-date使用，这一点在后面的快速参考中会再次出现。

SQL 资产中的注入：Jinja 模板

在 SQL 资产中，变量通过Jinja 模板注入到查询语句里。这是最常见的用法——用日期窗口做增量抽取或分区裁剪：

@bruin.asset(name="staging.monthly_trips", type="sql") 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(name="raw.monthly_data", type="python") 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 KEY=VALUE，值支持数组字面量：

bruin run ./pipeline.yml --var taxi_types=["green","fhv"]

也可以覆盖字符串类型的变量：

bruin run ./pipeline.yml --var customer_id=12345

在 Python 中读取自定义变量

与内置变量一样，自定义变量在 Python 中也以BRUIN_VAR_前缀的环境变量形式暴露。由于值是 JSON 编码的，数组类型需要用json.loads还原：

import os import json @bruin.asset(name="example.asset", type="python") 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）。面板主要提供三项能力：

1. Variable Override（变量覆盖）：运行前直接设置自定义变量的值，效果等同于 CLI 的--var；
2. Bruin Render（模板渲染预览）：实时查看 Jinja 模板被替换成真实值后的编译结果——这是排查日期窗口、引号、格式问题的最快途径；
3. Run Configuration（运行配置）：集中设置日期、环境（environment）与变量，一键发起运行。

对于「参数化管道 + 模板调试」的工作流，面板与 CLI 是互补的：面板适合开发期的即时预览与试运行，CLI 适合脚本化、定时化与 CI 中的正式执行。

典型应用场景

变量机制可以支撑以下几类常见的工程需求（整理自原文档）：

快速参考：常用运行命令

以下是本仓库笔记中整理的常用命令速查表，覆盖日期、变量与物化行为的控制：

# 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_id=12345 # Run with full refresh (affects materialization)（全量刷新，影响物化策略） bruin run ./pipeline.yml --full-refresh # Set end date as exclusive（end_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） → SQL（Jinja 注入）/ 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