企业官网建设流程全解析

使用 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-pipeline

bruin 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.db

05-data-platforms/notes/06-core-01-projects.md 对环境的用法给出了更完整的示例——通过default_environment指定默认生效的环境，可避免误在生产环境运行；同一连接类型可以配置多个实例，例如 default 环境用本地 DuckDB，production 环境指向 BigQuery：

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: production

Bruin 内置的连接类型包括 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

各字段含义：

值得强调的是连接作用域：连接定义在项目级（.bruin.yml），但每个管道显式声明自己使用哪些连接。这样大型组织中不同团队可以各用各的凭据、避免密钥过度暴露，Bruin 也只需为当前管道运行初始化必要的连接，实现团队间的安全隔离。

三类核心资产：Python、YAML ingestor 与 SQL

**资产（Asset）**是执行具体任务的单个文件，几乎总是与目标数据库中某张表/视图的创建或更新相关。每个资产文件包含两部分：定义（元数据、名称、类型、连接）与内容（实际执行的 SQL、Python 代码）。结合 05-data-platforms/notes/06-core-03-assets.md 的归纳，资产类型与典型用途如下：

资产名称可显式定义在装饰器中，也可默认从文件路径推断，因此建议按 schema/数据集组织目录：assets/raw/trips_raw.py对应建表raw.trips_raw，assets/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 内置的 ingestor（players.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_null

Seed 资产在运行结束后会自动执行声明的质量检查（如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 支持的物化策略汇总如下：

时间区间与增量摄取

Bruin 为每次运行注入内置变量start_date与end_date，取值由管道调度决定（05-data-platforms/notes/06-core-04-variables.md）：

使用方式有两种：

• 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 会：

1. 按依赖顺序运行资产——第一个资产完成后，自动触发下一个依赖它的资产；
2. 由全部依赖关系构建血缘图（lineage graph），供你查看与调试。

以 03 笔记中的三层管道为例，执行顺序是：摄取层（ingestion.trips与ingestion.payment_lookup并行）→ 暂存层staging.trips（依赖两个摄取资产都完成）→ 报表层reports.trips_report（依赖 staging 完成）。这一顺序完全由各资产depends声明推导，无需人工编排。在 VS Code/Cursor 的 Bruin 面板中打开 pipeline YAML 并切到 lineage 标签页，即可可视化全部资产及其依赖关系。

常用 CLI 命令速查

结合 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