使用 Bruin 快速上手数据平台安装、MCP 集成与第一个数据管道【免费下载链接】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 的 Getting Started 章节围绕数据平台 Bruin 的本地开发全流程展开安装 CLI 与 IDE 扩展、配置 Bruin MCP 让 AI Agent 协作建管道、用bruin init初始化项目、理解.bruin.yml与pipeline.yml的配置语义并掌握 Python、YAML ingestor、SQL 三类资产的编写方式。读完本文你将具备从零搭建一个带依赖、调度与增量摄取能力的 Bruin 数据管道项目的完整实战能力。Bruin 在数据平台中的定位在动手之前先明确 Bruin 解决什么问题。正如 05-data-platforms/notes/01-introduction.md 所述一个典型的数据栈需要多个组件协同从第三方源或数据库**摄取ingest数据、运行转换transform**任务、**编排orchestrate脚本的执行时机与依赖、并在交付前完成数据质量quality**校验。传统做法往往要用五六个独立工具分别配置而 Bruin 把代码逻辑、配置、依赖、质量检查、元数据与血缘整合进同一个平台、同一个项目里降低了搭建管道的工程门槛。05-data-platforms/README.md 将该模块定义为管理从摄取到分析全生命周期的数据平台实践Bruin 覆盖的能力包括数据摄取、数据转换、数据编排、内置质量检查以及血缘与文档等元数据管理。本篇聚焦于其中最基础的上手环节。安装 Bruin CLI 与 IDE 扩展安装命令行工具Bruin 的核心交互入口是 CLI。在终端中执行官方一键安装脚本curl -LsSf https://getbruin.com/install/cli | sh bruin version安装完成后用bruin version验证命令是否可用、确认版本号。后续所有操作初始化、校验、运行、查询都依赖该命令。安装 IDE 扩展安装 Bruin 的 VS Code 或 Cursor 扩展后编辑器会新增一个Bruin render panel渲染面板你可以在 IDE 内部直接运行资产asset和管道pipeline无需频繁切换终端面板还能展示资产的血缘关系与渲染后的实际 SQL 值详见后文变量注入一节。配置 Bruin MCP让 AI Agent 与你协作建管道MCP 是什么MCP 即Model Context Protocol模型上下文协议。Bruin 提供了 MCP 服务器使 Cursor、VS Code 等 IDE 中的 AI Agent 能够与 Bruin 通信——查询文档、代表你执行命令、阅读你的代码、排查报错、用自然语言分析数据。05-data-platforms/notes/04-bruin-mcp.md 总结了开启 MCP 后可获得的能力编写管道代码与资产配置编写文档与元数据排查错误与调试问题用自然语言执行查询、分析数据针对管道逻辑与结构提问VS Code 中的 MCP 配置mcp.json在仓库根目录与.git文件夹或package.json同级新建文件mcp.json粘贴以下配置{ servers: { bruin: { type: stdio, command: bruin, args: [ mcp ] } }, inputs: [] }这段配置指示 VS Code 启动bruin mcp命令通过标准输入/输出stdio与 Bruin MCP 服务器建立连接。若 IDE 显示连接失败可关闭并重新打开 IDE一般会看到 Bruin enabled 状态。从 05-data-platforms/notes/04-bruin-mcp.md 还可以看到 Cursor 与 Claude Code 的等价配置方式Cursor进入Settings → Tools MCP → New MCP Server添加{ mcpServers: { bruin: { command: bruin, args: [mcp] } } }Claude Code在终端执行claude mcp add bruin -- bruin mcp初始化第一个 Bruin 项目安装完成后用一条命令创建首个项目bruin init default my-first-pipeline cd my-first-pipelinebruin init会基于模板创建项目并自动完成三件事初始化 git 仓库、生成.gitignore、创建bruin.yaml即下文提到的.bruin.yml配置文件。Bruin 要求项目必须经过 git 初始化bruin init帮你自动处理了这一前置条件。提示本模块的实战课程使用的是 Zoomcamp 专用模板命令为bruin init zoomcamp my-taxi-pipeline它会生成一个基于 TODO 注释引导你逐步填写的练习项目本文使用的bruin init default模板同样遵循完全一致的项目结构。详见 05-data-platforms/README.md。项目结构解析bruin init default my-first-pipeline生成的项目结构如下my-first-pipeline/ ├── .bruin.yml # Environment and connection configuration ├── pipeline.yml # Pipeline name, schedule, default connections └── assets/ ├── players.asset.yml # Ingestr asset (data ingestion) ├── player_stats.sql # SQL asset with quality checks └── my_python_asset.py # Python asset三个核心元素项目级连接配置.bruin.yml、管道定义pipeline.yml、以及存放实际任务的assets/目录。.bruin.yml环境与连接配置这是项目的连接配置中心需要注意它的安全属性仅保存在本地bruin init已自动将其加入.gitignore绝不能推送到仓库——它包含数据库连接与密钥secrets定义环境default、production、staging 等每个环境下定义连接如 DuckDB、Chess.com API、自定义密钥最简配置示例default_environment: default environments: default: connections: duckdb: - name: duckdb-default path: duckdb.db05-data-platforms/notes/06-core-01-projects.md 对环境的用法给出了更完整的示例——通过default_environment指定默认生效的环境可避免误在生产环境运行同一连接类型可以配置多个实例例如 default 环境用本地 DuckDBproduction 环境指向 BigQuerydefault_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: productionBruin 内置的连接类型包括 DuckDB、MotherDuck、PostgreSQL、MySQL、BigQuery、Redshift、Snowflake以及用于存放 API Key 等凭据的自定义连接。多环境隔离带来的实际收益是本地开发、服务器部署可以共用同一套代码却不会暴露生产凭据不同团队也能按需分配不同的连接权限。pipeline.yml管道定义管道是按执行调度对资产进行分组的机制详见 05-data-platforms/notes/06-core-02-pipelines.md。pipeline.yml配置管道的名称、调度、默认连接与起始日期name: my-pipeline schedule: daily start_date: 2022-01-01 default_connections: duckdb: duckdb-default各字段含义字段说明name管道标识符schedule执行调度如hourly、daily、monthly或 cron 表达式一个管道只能有一个调度start_date管道开始生效的日期做全量刷新full refresh时从该日期开始处理数据default_connections该管道默认使用的连接在项目级.bruin.yml中定义此处按需引用variables管道级自定义变量详见下文时间区间与变量注入值得强调的是连接作用域连接定义在项目级.bruin.yml但每个管道显式声明自己使用哪些连接。这样大型组织中不同团队可以各用各的凭据、避免密钥过度暴露Bruin 也只需为当前管道运行初始化必要的连接实现团队间的安全隔离。三类核心资产Python、YAML ingestor 与 SQL**资产Asset**是执行具体任务的单个文件几乎总是与目标数据库中某张表/视图的创建或更新相关。每个资产文件包含两部分定义元数据、名称、类型、连接与内容实际执行的 SQL、Python 代码。结合 05-data-platforms/notes/06-core-03-assets.md 的归纳资产类型与典型用途如下类型描述典型场景PythonPython 脚本数据摄取、数据处理、ML 模型SQLSQL 查询转换、聚合YAML / Seed基于文件的表参考数据、静态查找表RR 脚本统计分析、R 专用工作流资产名称可显式定义在装饰器中也可默认从文件路径推断因此建议按 schema/数据集组织目录assets/raw/trips_raw.py对应建表raw.trips_rawassets/staging/trips_summary.sql对应建表staging.trips_summary。Python 资产最简单的形式一个带名称的 Python 脚本打印或处理数据可直接在 IDE 的 Bruin 面板中运行。摄取场景的完整形态可参考 05-data-platforms/notes/03-nyc-taxi-pipeline.md 中的trips.py——在文件头部的bruin ... bruin注释块中声明资产元数据与物化策略通过materialize()函数返回 DataFrame由 Bruin 负责写入目标表bruin name: ingestion.trips type: python image: python:3.11 materialization: type: table strategy: append columns: - name: pickup_datetime type: timestamp description: When the meter was engaged - name: dropoff_datetime type: timestamp description: When the meter was disengaged bruin 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]) # Generate list of months between start and end dates # Fetch parquet files from the NYC taxi trip-data endpoint return final_dataframe要点materialize()返回 DataFrame 后由 Bruin 完成落库append策略表示每次运行只插入新数据、不动已有行BRUIN_START_DATE/BRUIN_END_DATE提供时间窗口BRUIN_VARS可读取管道级自定义变量。YAML ingestor 资产利用 Bruin 内置的 ingestorplayers.asset.yml即此类。只需定义源连接、目标与表名无需手写摄取逻辑。它内置了大量源与目标类型Redshift、MySQL、Postgres、MotherDuck、BigQuery 等并且若目标数据库/表不存在会自动创建。Seed种子文件是 YAML 资产的另一常见形态用于把本地 CSV 载入数据库作为静态查找表例如 03 笔记中的支付方式映射表name: ingestion.payment_lookup type: duckdb.seed parameters: path: payment_lookup.csv columns: - name: payment_type_id type: integer description: Numeric code for payment type primary_key: true checks: - name: not_null - name: unique - name: payment_type_name type: string description: Human-readable payment type checks: - name: not_nullSeed 资产在运行结束后会自动执行声明的质量检查如not_null、unique。SQL 资产对数据库执行 SQL 查询完成转换/聚合。通过depends声明对其他资产的依赖——当依赖的资产运行完成后该资产自动触发。SQL 资产同样支持物化策略、列级检查与自定义检查完整示例同样可参考 03 笔记中的staging/trips.sql其核心结构为/* bruin name: staging.trips type: duckdb.sql depends: - ingestion.trips - ingestion.payment_lookup materialization: type: table strategy: time_interval incremental_key: pickup_datetime time_granularity: timestamp columns: - name: pickup_datetime type: timestamp primary_key: true checks: - name: not_null bruin */ SELECT ... FROM ingestion.trips t LEFT JOIN ingestion.payment_lookup p ON t.payment_type p.payment_type_id WHERE t.pickup_datetime {{ start_datetime }} AND t.pickup_datetime {{ end_datetime }}其中time_interval策略会先删除时间窗口内的行再插入查询结果因此 SQL 中的WHERE必须过滤到与运行区间一致的时间窗口以避免重复数据。物化策略速查结合 05-data-platforms/notes/03-nyc-taxi-pipeline.md 与 06-core-03 笔记Bruin 支持的物化策略汇总如下策略行为table每次运行删除并重建表view创建视图不存储数据append追加新数据不改动已有行merge基于键列做 upserttime_interval删除时间区间内行再重新插入deleteinsert删除匹配行后插入createreplace创建或替换表时间区间与增量摄取Bruin 为每次运行注入内置变量start_date与end_date取值由管道调度决定05-data-platforms/notes/06-core-04-variables.md调度start_dateend_dateMonthly当月第一天当月最后一天Daily当天开始当天结束Hourly当前小时开始当前小时结束使用方式有两种SQL 资产通过 Jinja 模板注入例如WHERE pickup_date {{ start_date }} AND pickup_date {{ end_date }}可在 VS Code 的 Bruin Render 面板查看编译后的真实值Python 资产通过环境变量读取如os.environ[BRUIN_VAR_START_DATE]自定义变量则使用BRUIN_VAR_前缀如BRUIN_VAR_TAXI_TYPES再以json.loads解析。内置的摄取ingestor资产会自动使用 start/end 日期因此你只需在运行或调度时设置start_date/end_date参数即可摄取指定时间区间的数据。此外管道级自定义变量如taxi_types可在pipeline.yml中定义默认值运行期用--var覆盖例如bruin run ./pipeline.yml --var taxi_types[green,fhv]从而支持日期分区、多租户处理、参数化转换与 A/B 测试等场景。依赖与数据血缘在资产中声明依赖后Bruin 会按依赖顺序运行资产——第一个资产完成后自动触发下一个依赖它的资产由全部依赖关系构建血缘图lineage graph供你查看与调试。以 03 笔记中的三层管道为例执行顺序是摄取层ingestion.trips与ingestion.payment_lookup并行→ 暂存层staging.trips依赖两个摄取资产都完成→ 报表层reports.trips_report依赖 staging 完成。这一顺序完全由各资产depends声明推导无需人工编排。在 VS Code/Cursor 的 Bruin 面板中打开 pipeline YAML 并切到 lineage 标签页即可可视化全部资产及其依赖关系。常用 CLI 命令速查命令用途bruin validate path检查语法与依赖不实际运行bruin run path执行管道或单个资产bruin run --downstream运行资产及其全部下游依赖bruin run --full-refresh清空并从头重建表bruin lineage path查看资产依赖关系bruin query --connection conn --query ...执行即席 SQL 查询结合 05-data-platforms/notes/06-core-05-commands.md几个高频组合用法# 运行前务必先校验检查循环依赖、资产定义、连接配置、引用完整性 bruin validate ./pipeline.yml # 指定时间区间运行适合小范围测试 bruin run ./pipeline.yml --start-date 2022-01-01 --end-date 2022-02-01 # 全量刷新并覆盖自定义变量 bruin run ./pipeline.yml --full-refresh --var taxi_types[yellow,green] # 只跑单个资产及其下游 bruin run ./pipeline.yml --asset raw.trips --downstream # 查看血缘 bruin lineage ./pipeline.yml # 即席查询结果 bruin query --connection duckdb-default --query SELECT COUNT(*) FROM staging.trips其中--exclusive-end-date可将结束日期设为开区间默认闭区间--environment指定运行环境dev/prod。一次bruin run会创建一个独立的run实例拥有自己的起止时间、变量值与执行日志。下一步从默认模板到完整管道本文展示的bruin init default my-first-pipeline项目已具备完整的数据平台骨架bruin validate保证配置正确bruin run按依赖执行三类资产bruin lineage呈现血缘bruin query验证结果。要继续深入可以参考本仓库模块内的两篇配套材料05-data-platforms/notes/03-nyc-taxi-pipeline.md基于 NYC Taxi 数据构建摄取 → 暂存 → 报表三层完整管道的端到端示例05-data-platforms/notes/04-bruin-mcp.md把本文配置的 MCP 用起来让 AI Agent 依据模板 README 中的提示词自动搭建整条管道并通过对话式查询验证数据。若准备将管道投入生产05-data-platforms/notes/05-bruin-cloud.md 介绍了如何把本地项目部署到 Bruin Cloud实现托管调度与监控。【免费下载链接】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 快速上手数据平台：安装、MCP 集成与第一个数据管道
使用 Bruin 快速上手数据平台安装、MCP 集成与第一个数据管道【免费下载链接】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 的 Getting Started 章节围绕数据平台 Bruin 的本地开发全流程展开安装 CLI 与 IDE 扩展、配置 Bruin MCP 让 AI Agent 协作建管道、用bruin init初始化项目、理解.bruin.yml与pipeline.yml的配置语义并掌握 Python、YAML ingestor、SQL 三类资产的编写方式。读完本文你将具备从零搭建一个带依赖、调度与增量摄取能力的 Bruin 数据管道项目的完整实战能力。Bruin 在数据平台中的定位在动手之前先明确 Bruin 解决什么问题。正如 05-data-platforms/notes/01-introduction.md 所述一个典型的数据栈需要多个组件协同从第三方源或数据库**摄取ingest数据、运行转换transform**任务、**编排orchestrate脚本的执行时机与依赖、并在交付前完成数据质量quality**校验。传统做法往往要用五六个独立工具分别配置而 Bruin 把代码逻辑、配置、依赖、质量检查、元数据与血缘整合进同一个平台、同一个项目里降低了搭建管道的工程门槛。05-data-platforms/README.md 将该模块定义为管理从摄取到分析全生命周期的数据平台实践Bruin 覆盖的能力包括数据摄取、数据转换、数据编排、内置质量检查以及血缘与文档等元数据管理。本篇聚焦于其中最基础的上手环节。安装 Bruin CLI 与 IDE 扩展安装命令行工具Bruin 的核心交互入口是 CLI。在终端中执行官方一键安装脚本curl -LsSf https://getbruin.com/install/cli | sh bruin version安装完成后用bruin version验证命令是否可用、确认版本号。后续所有操作初始化、校验、运行、查询都依赖该命令。安装 IDE 扩展安装 Bruin 的 VS Code 或 Cursor 扩展后编辑器会新增一个Bruin render panel渲染面板你可以在 IDE 内部直接运行资产asset和管道pipeline无需频繁切换终端面板还能展示资产的血缘关系与渲染后的实际 SQL 值详见后文变量注入一节。配置 Bruin MCP让 AI Agent 与你协作建管道MCP 是什么MCP 即Model Context Protocol模型上下文协议。Bruin 提供了 MCP 服务器使 Cursor、VS Code 等 IDE 中的 AI Agent 能够与 Bruin 通信——查询文档、代表你执行命令、阅读你的代码、排查报错、用自然语言分析数据。05-data-platforms/notes/04-bruin-mcp.md 总结了开启 MCP 后可获得的能力编写管道代码与资产配置编写文档与元数据排查错误与调试问题用自然语言执行查询、分析数据针对管道逻辑与结构提问VS Code 中的 MCP 配置mcp.json在仓库根目录与.git文件夹或package.json同级新建文件mcp.json粘贴以下配置{ servers: { bruin: { type: stdio, command: bruin, args: [ mcp ] } }, inputs: [] }这段配置指示 VS Code 启动bruin mcp命令通过标准输入/输出stdio与 Bruin MCP 服务器建立连接。若 IDE 显示连接失败可关闭并重新打开 IDE一般会看到 Bruin enabled 状态。从 05-data-platforms/notes/04-bruin-mcp.md 还可以看到 Cursor 与 Claude Code 的等价配置方式Cursor进入Settings → Tools MCP → New MCP Server添加{ mcpServers: { bruin: { command: bruin, args: [mcp] } } }Claude Code在终端执行claude mcp add bruin -- bruin mcp初始化第一个 Bruin 项目安装完成后用一条命令创建首个项目bruin init default my-first-pipeline cd my-first-pipelinebruin init会基于模板创建项目并自动完成三件事初始化 git 仓库、生成.gitignore、创建bruin.yaml即下文提到的.bruin.yml配置文件。Bruin 要求项目必须经过 git 初始化bruin init帮你自动处理了这一前置条件。提示本模块的实战课程使用的是 Zoomcamp 专用模板命令为bruin init zoomcamp my-taxi-pipeline它会生成一个基于 TODO 注释引导你逐步填写的练习项目本文使用的bruin init default模板同样遵循完全一致的项目结构。详见 05-data-platforms/README.md。项目结构解析bruin init default my-first-pipeline生成的项目结构如下my-first-pipeline/ ├── .bruin.yml # Environment and connection configuration ├── pipeline.yml # Pipeline name, schedule, default connections └── assets/ ├── players.asset.yml # Ingestr asset (data ingestion) ├── player_stats.sql # SQL asset with quality checks └── my_python_asset.py # Python asset三个核心元素项目级连接配置.bruin.yml、管道定义pipeline.yml、以及存放实际任务的assets/目录。.bruin.yml环境与连接配置这是项目的连接配置中心需要注意它的安全属性仅保存在本地bruin init已自动将其加入.gitignore绝不能推送到仓库——它包含数据库连接与密钥secrets定义环境default、production、staging 等每个环境下定义连接如 DuckDB、Chess.com API、自定义密钥最简配置示例default_environment: default environments: default: connections: duckdb: - name: duckdb-default path: duckdb.db05-data-platforms/notes/06-core-01-projects.md 对环境的用法给出了更完整的示例——通过default_environment指定默认生效的环境可避免误在生产环境运行同一连接类型可以配置多个实例例如 default 环境用本地 DuckDBproduction 环境指向 BigQuerydefault_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: productionBruin 内置的连接类型包括 DuckDB、MotherDuck、PostgreSQL、MySQL、BigQuery、Redshift、Snowflake以及用于存放 API Key 等凭据的自定义连接。多环境隔离带来的实际收益是本地开发、服务器部署可以共用同一套代码却不会暴露生产凭据不同团队也能按需分配不同的连接权限。pipeline.yml管道定义管道是按执行调度对资产进行分组的机制详见 05-data-platforms/notes/06-core-02-pipelines.md。pipeline.yml配置管道的名称、调度、默认连接与起始日期name: my-pipeline schedule: daily start_date: 2022-01-01 default_connections: duckdb: duckdb-default各字段含义字段说明name管道标识符schedule执行调度如hourly、daily、monthly或 cron 表达式一个管道只能有一个调度start_date管道开始生效的日期做全量刷新full refresh时从该日期开始处理数据default_connections该管道默认使用的连接在项目级.bruin.yml中定义此处按需引用variables管道级自定义变量详见下文时间区间与变量注入值得强调的是连接作用域连接定义在项目级.bruin.yml但每个管道显式声明自己使用哪些连接。这样大型组织中不同团队可以各用各的凭据、避免密钥过度暴露Bruin 也只需为当前管道运行初始化必要的连接实现团队间的安全隔离。三类核心资产Python、YAML ingestor 与 SQL**资产Asset**是执行具体任务的单个文件几乎总是与目标数据库中某张表/视图的创建或更新相关。每个资产文件包含两部分定义元数据、名称、类型、连接与内容实际执行的 SQL、Python 代码。结合 05-data-platforms/notes/06-core-03-assets.md 的归纳资产类型与典型用途如下类型描述典型场景PythonPython 脚本数据摄取、数据处理、ML 模型SQLSQL 查询转换、聚合YAML / Seed基于文件的表参考数据、静态查找表RR 脚本统计分析、R 专用工作流资产名称可显式定义在装饰器中也可默认从文件路径推断因此建议按 schema/数据集组织目录assets/raw/trips_raw.py对应建表raw.trips_rawassets/staging/trips_summary.sql对应建表staging.trips_summary。Python 资产最简单的形式一个带名称的 Python 脚本打印或处理数据可直接在 IDE 的 Bruin 面板中运行。摄取场景的完整形态可参考 05-data-platforms/notes/03-nyc-taxi-pipeline.md 中的trips.py——在文件头部的bruin ... bruin注释块中声明资产元数据与物化策略通过materialize()函数返回 DataFrame由 Bruin 负责写入目标表bruin name: ingestion.trips type: python image: python:3.11 materialization: type: table strategy: append columns: - name: pickup_datetime type: timestamp description: When the meter was engaged - name: dropoff_datetime type: timestamp description: When the meter was disengaged bruin 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]) # Generate list of months between start and end dates # Fetch parquet files from the NYC taxi trip-data endpoint return final_dataframe要点materialize()返回 DataFrame 后由 Bruin 完成落库append策略表示每次运行只插入新数据、不动已有行BRUIN_START_DATE/BRUIN_END_DATE提供时间窗口BRUIN_VARS可读取管道级自定义变量。YAML ingestor 资产利用 Bruin 内置的 ingestorplayers.asset.yml即此类。只需定义源连接、目标与表名无需手写摄取逻辑。它内置了大量源与目标类型Redshift、MySQL、Postgres、MotherDuck、BigQuery 等并且若目标数据库/表不存在会自动创建。Seed种子文件是 YAML 资产的另一常见形态用于把本地 CSV 载入数据库作为静态查找表例如 03 笔记中的支付方式映射表name: ingestion.payment_lookup type: duckdb.seed parameters: path: payment_lookup.csv columns: - name: payment_type_id type: integer description: Numeric code for payment type primary_key: true checks: - name: not_null - name: unique - name: payment_type_name type: string description: Human-readable payment type checks: - name: not_nullSeed 资产在运行结束后会自动执行声明的质量检查如not_null、unique。SQL 资产对数据库执行 SQL 查询完成转换/聚合。通过depends声明对其他资产的依赖——当依赖的资产运行完成后该资产自动触发。SQL 资产同样支持物化策略、列级检查与自定义检查完整示例同样可参考 03 笔记中的staging/trips.sql其核心结构为/* bruin name: staging.trips type: duckdb.sql depends: - ingestion.trips - ingestion.payment_lookup materialization: type: table strategy: time_interval incremental_key: pickup_datetime time_granularity: timestamp columns: - name: pickup_datetime type: timestamp primary_key: true checks: - name: not_null bruin */ SELECT ... FROM ingestion.trips t LEFT JOIN ingestion.payment_lookup p ON t.payment_type p.payment_type_id WHERE t.pickup_datetime {{ start_datetime }} AND t.pickup_datetime {{ end_datetime }}其中time_interval策略会先删除时间窗口内的行再插入查询结果因此 SQL 中的WHERE必须过滤到与运行区间一致的时间窗口以避免重复数据。物化策略速查结合 05-data-platforms/notes/03-nyc-taxi-pipeline.md 与 06-core-03 笔记Bruin 支持的物化策略汇总如下策略行为table每次运行删除并重建表view创建视图不存储数据append追加新数据不改动已有行merge基于键列做 upserttime_interval删除时间区间内行再重新插入deleteinsert删除匹配行后插入createreplace创建或替换表时间区间与增量摄取Bruin 为每次运行注入内置变量start_date与end_date取值由管道调度决定05-data-platforms/notes/06-core-04-variables.md调度start_dateend_dateMonthly当月第一天当月最后一天Daily当天开始当天结束Hourly当前小时开始当前小时结束使用方式有两种SQL 资产通过 Jinja 模板注入例如WHERE pickup_date {{ start_date }} AND pickup_date {{ end_date }}可在 VS Code 的 Bruin Render 面板查看编译后的真实值Python 资产通过环境变量读取如os.environ[BRUIN_VAR_START_DATE]自定义变量则使用BRUIN_VAR_前缀如BRUIN_VAR_TAXI_TYPES再以json.loads解析。内置的摄取ingestor资产会自动使用 start/end 日期因此你只需在运行或调度时设置start_date/end_date参数即可摄取指定时间区间的数据。此外管道级自定义变量如taxi_types可在pipeline.yml中定义默认值运行期用--var覆盖例如bruin run ./pipeline.yml --var taxi_types[green,fhv]从而支持日期分区、多租户处理、参数化转换与 A/B 测试等场景。依赖与数据血缘在资产中声明依赖后Bruin 会按依赖顺序运行资产——第一个资产完成后自动触发下一个依赖它的资产由全部依赖关系构建血缘图lineage graph供你查看与调试。以 03 笔记中的三层管道为例执行顺序是摄取层ingestion.trips与ingestion.payment_lookup并行→ 暂存层staging.trips依赖两个摄取资产都完成→ 报表层reports.trips_report依赖 staging 完成。这一顺序完全由各资产depends声明推导无需人工编排。在 VS Code/Cursor 的 Bruin 面板中打开 pipeline YAML 并切到 lineage 标签页即可可视化全部资产及其依赖关系。常用 CLI 命令速查命令用途bruin validate path检查语法与依赖不实际运行bruin run path执行管道或单个资产bruin run --downstream运行资产及其全部下游依赖bruin run --full-refresh清空并从头重建表bruin lineage path查看资产依赖关系bruin query --connection conn --query ...执行即席 SQL 查询结合 05-data-platforms/notes/06-core-05-commands.md几个高频组合用法# 运行前务必先校验检查循环依赖、资产定义、连接配置、引用完整性 bruin validate ./pipeline.yml # 指定时间区间运行适合小范围测试 bruin run ./pipeline.yml --start-date 2022-01-01 --end-date 2022-02-01 # 全量刷新并覆盖自定义变量 bruin run ./pipeline.yml --full-refresh --var taxi_types[yellow,green] # 只跑单个资产及其下游 bruin run ./pipeline.yml --asset raw.trips --downstream # 查看血缘 bruin lineage ./pipeline.yml # 即席查询结果 bruin query --connection duckdb-default --query SELECT COUNT(*) FROM staging.trips其中--exclusive-end-date可将结束日期设为开区间默认闭区间--environment指定运行环境dev/prod。一次bruin run会创建一个独立的run实例拥有自己的起止时间、变量值与执行日志。下一步从默认模板到完整管道本文展示的bruin init default my-first-pipeline项目已具备完整的数据平台骨架bruin validate保证配置正确bruin run按依赖执行三类资产bruin lineage呈现血缘bruin query验证结果。要继续深入可以参考本仓库模块内的两篇配套材料05-data-platforms/notes/03-nyc-taxi-pipeline.md基于 NYC Taxi 数据构建摄取 → 暂存 → 报表三层完整管道的端到端示例05-data-platforms/notes/04-bruin-mcp.md把本文配置的 MCP 用起来让 AI Agent 依据模板 README 中的提示词自动搭建整条管道并通过对话式查询验证数据。若准备将管道投入生产05-data-platforms/notes/05-bruin-cloud.md 介绍了如何把本地项目部署到 Bruin Cloud实现托管调度与监控。【免费下载链接】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），仅供参考