Bruin Pipeline 核心概念深度解析调度分组、pipeline.yml 配置与连接作用域Data Engineering Zoomcamp 2027【免费下载链接】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 2027 课程模块 05-data-platforms 中“Core Concepts: Pipelines”讲义为骨架系统讲解 Bruin 数据平台中 Pipeline 的定义、目录结构、pipeline.yml配置与连接作用域机制并结合同一课程模块中的 Project、Asset、Variable、Commands 讲义与 NYC Taxi 三层管线实战示例进行源码级交叉印证。读完本文你将掌握如何按调度与配置合理拆分 Pipeline、编写可复用的pipeline.yml、隔离不同团队的数据库凭据并用bruin validate/bruin lineage/bruin run完成流水线的验证、血缘查看与执行。Pipeline 是什么按调度与配置分组的数据资产编排单元在 Bruin 中Pipeline流水线是一种分组机制grouping mechanism用于把数据资产Asset按照执行调度和配置需求组织在一起。在一个 Project 内可以同时存在多个 Pipeline。讲义中给出的精确定义是A pipeline is the operational grouping: shared schedule and connection scope around a dependency graph of assets.Pipeline 是操作层面的分组围绕一张资产依赖图共享同一套调度与连接作用域。要理解这个定义需要把它放回 Bruin 的分层模型中看待。结合本模块的姊妹讲义 Core Concepts: Projects 和 Core Concepts: Assets 可以看出Project项目仓库根目录是配置的边界存放.bruin.yml环境与连接定义Pipeline流水线处于 Project 之下、Asset 之上的中间层负责把“什么时间跑、用什么连接”这类操作属性集中起来Asset资产真正干活的单个文件SQL、Python、Seed 等创建或更新目标数据库中的表/视图。Pipeline 并不重复定义资产本身而是为资产图提供两个关键的操作属性共享调度与连接作用域。这正是它与 Project、Asset 之间职责划分的要点。为什么需要 Pipeline单一调度是分组的第一动机讲义明确指出每个 Pipeline只有一个调度one schedule这也是把资产分到同一组的最主要原因调度相同的资产应该放进同一个 Pipeline常用调度包括hourly每小时、daily每天、monthly每月以及任意cron 表达式。“一个 Pipeline 一个调度”意味着如果你有一批任务每天跑、另一批任务每月跑就应该把它们拆成两个 Pipeline而不是塞进同一个流水线里用一个调度硬套。这种约束看似简单却能保证每个 Pipeline 的时间语义是自洽的——因为 Bruin 会根据 Pipeline 的调度自动计算内置变量start_date/end_date详见 Core Concepts: Variables同一 Pipeline 内的所有资产共享同一时间窗口避免出现“日调度任务里混着月窗口计算”这类逻辑错乱。在本模块的实战讲义 Building an End-to-End Pipeline with NYC Taxi Data 中NYC Taxi 示例就是把 ingestion摄取、staging清洗、reports报表三层资产统一放进一个schedule: daily的 Pipeline 中让三层资产共享同一日调度与同一 DuckDB 连接。Pipeline 的目录结构每个 Pipeline 在项目内拥有自己的文件夹其中包含一个pipeline.yml文件。讲义给出的标准结构如下project/ ├── .bruin.yml ├── pipelines/ │ ├── nyc-taxi/ │ │ ├── pipeline.yml │ │ └── assets/ │ └── another-pipeline/ │ ├── pipeline.yml │ └── assets/要点pipeline.yml是每个 Pipeline 的“身份证”定义名称、调度、起始日期、默认连接与变量assets/目录存放该 Pipeline 专属的数据资产SQL / Python / Seed 文件不同 Pipeline 之间可以通过依赖关系协同也可以完全独立。与 Getting Started with Bruin 中展示的最小化项目结构pipeline.ymlassets/平铺在项目根下相比这里的pipelines/目录是多 Pipeline 项目的推荐组织方式每个业务域一个子目录目录即边界。深入pipeline.yml字段与配置说明pipeline.yml是 Pipeline 的核心配置文件。讲义给出的最小示例name: nyc_taxi schedule: monthly start_date: 2019-01-01 default_connections: duckdb: duckdb-default配置项速查表SettingDescription含义namePipeline 标识符流水线名称schedule何时运行cron、daily、monthly 等调度表达式start_datePipeline 何时开始生效回溯执行的起点default_connections该 Pipeline 默认使用哪些连接variables该 Pipeline 的自定义变量对照实战NYC Taxi 的pipeline.yml在 03-nyc-taxi-pipeline.md 中完整的pipeline.yml展示了这些字段如何协同工作name: nyc_taxi schedule: daily start_date: 2022-01-01 default_connections: duckdb: duckdb-default variables: taxi_types: type: array items: type: string default: [yellow]对照说明schedule: daily整个 Pipeline 按日调度同一 Pipeline 内的 ingestion / staging / reports 三层资产都共享这个日窗口start_date讲义特别注明做--full-refresh全量刷新时Bruin 会从这个日期开始回溯处理数据因此它既是“生效日期”也是“回溯起点”default_connections声明本 Pipeline 需要duckdb-default连接variables定义了自定义变量taxi_types数组类型默认[yellow]用来控制摄取 yellow 还是 green 出租车数据并可在运行时用--var覆盖。关于schedule的取值讲义说明schedule支持 cron 表达式以及hourly、daily、monthly等常见调度。其中start_date/end_date内置变量会随调度粒度变化——monthly 对应月初至月末、daily 对应当日 0 点至 24 点、hourly 对应整点区间详见 Core Concepts: Variables。因此选择调度时实际也在定义整个 Pipeline 的时间语义粒度。连接作用域Connection Scoping安全与隔离的关键设计讲义强调了一个容易被忽略但极其重要的设计连接定义在项目层.bruin.yml但每个 Pipeline 必须显式声明自己使用哪些连接即default_connections。为什么这样做讲义列出的理由在大型组织中不同团队可能需要不同的凭据防止不必要的密钥secrets暴露只初始化特定 Pipeline 运行所需的连接避免多余的连接初始化开销在部门之间实现安全隔离。项目层的连接定义示例连接的具体配置位于项目根目录的.bruin.yml该文件会自动加入.gitignore只保存在本地绝不能推送到仓库来自 Core Concepts: Projects 的示例default_environment: default environments: default: connections: duckdb: - name: duckdb-default path: duckdb.db motherduck: - name: motherduck token: your-token production: connections: bigquery: - name: bq-prod project: my-project dataset: productionPipeline 层如何引用Pipeline 不重复写连接细节只写引用default_connections: duckdb: duckdb-default这形成了一个清晰的“定义在项目、作用在流水线”的两层模型.bruin.yml是连接定义的唯一事实来源pipeline.yml只声明“我这个流水线需要哪几个连接”。配合环境default_environment设为dev/default防止误跑生产连接作用域在凭据安全与多环境隔离上起到了双重保障作用。与变量配合Pipeline 级参数化pipeline.yml中的variables字段见上文配置项速查表是 Pipeline 级参数化能力的入口。结合 Core Concepts: Variables 可以梳理出完整的用法内置变量start_date、end_date由 Pipeline 的调度自动计算SQL 资产通过 Jinja 模板{{ start_date }}、{{ end_date }}注入Python 资产通过环境变量BRUIN_VAR_START_DATE/BRUIN_VAR_END_DATE读取自定义变量在pipeline.yml中声明默认值运行时通过--var覆盖bruin run ./pipelines/nyc-taxi/pipeline.yml --var taxi_types[green,fhv]这意味着同一个 Pipeline 可以在不改动任何资产逻辑的前提下为每次运行选择不同的时间区间和自定义参数——多租户处理、按日期分区抽取、A/B 测试等场景都依赖这一机制。命令速查验证、血缘与运行讲义为 Pipeline 提供了三个最常用的 CLI 命令完整命令体系见 Core Concepts: Commands# 验证 Pipeline 配置检查循环依赖、资产定义、连接有效性 bruin validate ./pipelines/nyc-taxi/pipeline.yml # 查看 Pipeline 的血缘资产依赖图 bruin lineage ./pipelines/nyc-taxi/pipeline.yml # 运行整个 Pipeline按依赖顺序执行全部资产 bruin run ./pipelines/nyc-taxi/pipeline.yml其中bruin validate在运行前检查血缘中是否存在循环依赖、资产定义是否正确、连接是否存在且配置无误、引用是否断裂——讲义强调“Always validate before running!”bruin lineage可视化资产之间的上下游关系帮助确认执行顺序是否符合预期bruin run创建一次独立的执行实例run可配合--start-date、--end-date、--full-refresh、--environment、--var、--asset、--upstream、--downstream等参数精确控制执行范围。整体协作Pipeline 在 Bruin 工作流中的位置结合 Core Concepts: Commands 给出的完整工作流可以把 Pipeline 放到整个 Bruin 体系中定位1. Project项目根目录已初始化 └── .bruin.yml环境、连接 2. Pipeline按调度分组的流水线 └── pipeline.yml调度、默认连接、变量 3. Assets真正执行的工作 ├── Python摄取、处理 ├── SQL转换 └── YAML/Seed静态数据 4. Commands驱动一切 ├── bruin run执行 ├── bruin validate校验 └── bruin query查询在 NYC Taxi 实战 中这种分层协作的具体执行顺序为ingestion 层资产trips.pypayment_lookup种子文件并行先行staging 层 SQL 资产在两者完成后运行通过depends声明依赖reports 层 SQL 资产最后运行。Pipeline 的default_connections与调度把这三层粘合成一个可按日调度的整体。进一步阅读本模块 README 与课程索引Core Concepts: Projects项目、.bruin.yml与环境Core Concepts: Assets资产类型、物化策略与依赖Core Concepts: Variables内置/自定义变量与运行时覆盖Core Concepts: Commandsrun / validate / lineage / query 全命令参考Building an End-to-End Pipeline with NYC Taxi Data完整三层管线实战Getting Started with Bruin安装、初始化与最小项目结构【免费下载链接】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 Pipeline 核心概念深度解析：调度分组、pipeline.yml 配置与连接作用域（Data Engineering Zoomcamp 2027）
Bruin Pipeline 核心概念深度解析调度分组、pipeline.yml 配置与连接作用域Data Engineering Zoomcamp 2027【免费下载链接】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 2027 课程模块 05-data-platforms 中“Core Concepts: Pipelines”讲义为骨架系统讲解 Bruin 数据平台中 Pipeline 的定义、目录结构、pipeline.yml配置与连接作用域机制并结合同一课程模块中的 Project、Asset、Variable、Commands 讲义与 NYC Taxi 三层管线实战示例进行源码级交叉印证。读完本文你将掌握如何按调度与配置合理拆分 Pipeline、编写可复用的pipeline.yml、隔离不同团队的数据库凭据并用bruin validate/bruin lineage/bruin run完成流水线的验证、血缘查看与执行。Pipeline 是什么按调度与配置分组的数据资产编排单元在 Bruin 中Pipeline流水线是一种分组机制grouping mechanism用于把数据资产Asset按照执行调度和配置需求组织在一起。在一个 Project 内可以同时存在多个 Pipeline。讲义中给出的精确定义是A pipeline is the operational grouping: shared schedule and connection scope around a dependency graph of assets.Pipeline 是操作层面的分组围绕一张资产依赖图共享同一套调度与连接作用域。要理解这个定义需要把它放回 Bruin 的分层模型中看待。结合本模块的姊妹讲义 Core Concepts: Projects 和 Core Concepts: Assets 可以看出Project项目仓库根目录是配置的边界存放.bruin.yml环境与连接定义Pipeline流水线处于 Project 之下、Asset 之上的中间层负责把“什么时间跑、用什么连接”这类操作属性集中起来Asset资产真正干活的单个文件SQL、Python、Seed 等创建或更新目标数据库中的表/视图。Pipeline 并不重复定义资产本身而是为资产图提供两个关键的操作属性共享调度与连接作用域。这正是它与 Project、Asset 之间职责划分的要点。为什么需要 Pipeline单一调度是分组的第一动机讲义明确指出每个 Pipeline只有一个调度one schedule这也是把资产分到同一组的最主要原因调度相同的资产应该放进同一个 Pipeline常用调度包括hourly每小时、daily每天、monthly每月以及任意cron 表达式。“一个 Pipeline 一个调度”意味着如果你有一批任务每天跑、另一批任务每月跑就应该把它们拆成两个 Pipeline而不是塞进同一个流水线里用一个调度硬套。这种约束看似简单却能保证每个 Pipeline 的时间语义是自洽的——因为 Bruin 会根据 Pipeline 的调度自动计算内置变量start_date/end_date详见 Core Concepts: Variables同一 Pipeline 内的所有资产共享同一时间窗口避免出现“日调度任务里混着月窗口计算”这类逻辑错乱。在本模块的实战讲义 Building an End-to-End Pipeline with NYC Taxi Data 中NYC Taxi 示例就是把 ingestion摄取、staging清洗、reports报表三层资产统一放进一个schedule: daily的 Pipeline 中让三层资产共享同一日调度与同一 DuckDB 连接。Pipeline 的目录结构每个 Pipeline 在项目内拥有自己的文件夹其中包含一个pipeline.yml文件。讲义给出的标准结构如下project/ ├── .bruin.yml ├── pipelines/ │ ├── nyc-taxi/ │ │ ├── pipeline.yml │ │ └── assets/ │ └── another-pipeline/ │ ├── pipeline.yml │ └── assets/要点pipeline.yml是每个 Pipeline 的“身份证”定义名称、调度、起始日期、默认连接与变量assets/目录存放该 Pipeline 专属的数据资产SQL / Python / Seed 文件不同 Pipeline 之间可以通过依赖关系协同也可以完全独立。与 Getting Started with Bruin 中展示的最小化项目结构pipeline.ymlassets/平铺在项目根下相比这里的pipelines/目录是多 Pipeline 项目的推荐组织方式每个业务域一个子目录目录即边界。深入pipeline.yml字段与配置说明pipeline.yml是 Pipeline 的核心配置文件。讲义给出的最小示例name: nyc_taxi schedule: monthly start_date: 2019-01-01 default_connections: duckdb: duckdb-default配置项速查表SettingDescription含义namePipeline 标识符流水线名称schedule何时运行cron、daily、monthly 等调度表达式start_datePipeline 何时开始生效回溯执行的起点default_connections该 Pipeline 默认使用哪些连接variables该 Pipeline 的自定义变量对照实战NYC Taxi 的pipeline.yml在 03-nyc-taxi-pipeline.md 中完整的pipeline.yml展示了这些字段如何协同工作name: nyc_taxi schedule: daily start_date: 2022-01-01 default_connections: duckdb: duckdb-default variables: taxi_types: type: array items: type: string default: [yellow]对照说明schedule: daily整个 Pipeline 按日调度同一 Pipeline 内的 ingestion / staging / reports 三层资产都共享这个日窗口start_date讲义特别注明做--full-refresh全量刷新时Bruin 会从这个日期开始回溯处理数据因此它既是“生效日期”也是“回溯起点”default_connections声明本 Pipeline 需要duckdb-default连接variables定义了自定义变量taxi_types数组类型默认[yellow]用来控制摄取 yellow 还是 green 出租车数据并可在运行时用--var覆盖。关于schedule的取值讲义说明schedule支持 cron 表达式以及hourly、daily、monthly等常见调度。其中start_date/end_date内置变量会随调度粒度变化——monthly 对应月初至月末、daily 对应当日 0 点至 24 点、hourly 对应整点区间详见 Core Concepts: Variables。因此选择调度时实际也在定义整个 Pipeline 的时间语义粒度。连接作用域Connection Scoping安全与隔离的关键设计讲义强调了一个容易被忽略但极其重要的设计连接定义在项目层.bruin.yml但每个 Pipeline 必须显式声明自己使用哪些连接即default_connections。为什么这样做讲义列出的理由在大型组织中不同团队可能需要不同的凭据防止不必要的密钥secrets暴露只初始化特定 Pipeline 运行所需的连接避免多余的连接初始化开销在部门之间实现安全隔离。项目层的连接定义示例连接的具体配置位于项目根目录的.bruin.yml该文件会自动加入.gitignore只保存在本地绝不能推送到仓库来自 Core Concepts: Projects 的示例default_environment: default environments: default: connections: duckdb: - name: duckdb-default path: duckdb.db motherduck: - name: motherduck token: your-token production: connections: bigquery: - name: bq-prod project: my-project dataset: productionPipeline 层如何引用Pipeline 不重复写连接细节只写引用default_connections: duckdb: duckdb-default这形成了一个清晰的“定义在项目、作用在流水线”的两层模型.bruin.yml是连接定义的唯一事实来源pipeline.yml只声明“我这个流水线需要哪几个连接”。配合环境default_environment设为dev/default防止误跑生产连接作用域在凭据安全与多环境隔离上起到了双重保障作用。与变量配合Pipeline 级参数化pipeline.yml中的variables字段见上文配置项速查表是 Pipeline 级参数化能力的入口。结合 Core Concepts: Variables 可以梳理出完整的用法内置变量start_date、end_date由 Pipeline 的调度自动计算SQL 资产通过 Jinja 模板{{ start_date }}、{{ end_date }}注入Python 资产通过环境变量BRUIN_VAR_START_DATE/BRUIN_VAR_END_DATE读取自定义变量在pipeline.yml中声明默认值运行时通过--var覆盖bruin run ./pipelines/nyc-taxi/pipeline.yml --var taxi_types[green,fhv]这意味着同一个 Pipeline 可以在不改动任何资产逻辑的前提下为每次运行选择不同的时间区间和自定义参数——多租户处理、按日期分区抽取、A/B 测试等场景都依赖这一机制。命令速查验证、血缘与运行讲义为 Pipeline 提供了三个最常用的 CLI 命令完整命令体系见 Core Concepts: Commands# 验证 Pipeline 配置检查循环依赖、资产定义、连接有效性 bruin validate ./pipelines/nyc-taxi/pipeline.yml # 查看 Pipeline 的血缘资产依赖图 bruin lineage ./pipelines/nyc-taxi/pipeline.yml # 运行整个 Pipeline按依赖顺序执行全部资产 bruin run ./pipelines/nyc-taxi/pipeline.yml其中bruin validate在运行前检查血缘中是否存在循环依赖、资产定义是否正确、连接是否存在且配置无误、引用是否断裂——讲义强调“Always validate before running!”bruin lineage可视化资产之间的上下游关系帮助确认执行顺序是否符合预期bruin run创建一次独立的执行实例run可配合--start-date、--end-date、--full-refresh、--environment、--var、--asset、--upstream、--downstream等参数精确控制执行范围。整体协作Pipeline 在 Bruin 工作流中的位置结合 Core Concepts: Commands 给出的完整工作流可以把 Pipeline 放到整个 Bruin 体系中定位1. Project项目根目录已初始化 └── .bruin.yml环境、连接 2. Pipeline按调度分组的流水线 └── pipeline.yml调度、默认连接、变量 3. Assets真正执行的工作 ├── Python摄取、处理 ├── SQL转换 └── YAML/Seed静态数据 4. Commands驱动一切 ├── bruin run执行 ├── bruin validate校验 └── bruin query查询在 NYC Taxi 实战 中这种分层协作的具体执行顺序为ingestion 层资产trips.pypayment_lookup种子文件并行先行staging 层 SQL 资产在两者完成后运行通过depends声明依赖reports 层 SQL 资产最后运行。Pipeline 的default_connections与调度把这三层粘合成一个可按日调度的整体。进一步阅读本模块 README 与课程索引Core Concepts: Projects项目、.bruin.yml与环境Core Concepts: Assets资产类型、物化策略与依赖Core Concepts: Variables内置/自定义变量与运行时覆盖Core Concepts: Commandsrun / validate / lineage / query 全命令参考Building an End-to-End Pipeline with NYC Taxi Data完整三层管线实战Getting Started with Bruin安装、初始化与最小项目结构【免费下载链接】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），仅供参考