De 3 4 5 Bigquery Partitioning Gcp Buckets Dbt Bruin MCP

zh-3-4-5-bigquery-partitioning-gcp-buckets-dbt-duckdb-brown-mcp

用于GCP、DBT、数据工程平台、批处理和流媒体的各种技术沙箱。

概述

该项目将纽约市黄色出租车行程数据从TLC公共数据集中加载到谷歌云存储（GCS）存储桶中。数据以Parquet格式下载，并同时上传以进行高效处理。

设置

1. 克隆存储库：

   git clone 
   cd de-3-4-5-bigquery-partitioning-gcp-buckets-dbt-bruin-mcp

1. 使用uv安装依赖项：

   uv sync

1. 配置GCP凭据：

- 将您的GCP服务帐户JSON密钥文件放置为 gcs.json 在项目根中 - 更新 BUCKET_NAME 在 load_yellow_taxi_data.py 如有需要

用法

运行数据加载脚本：

python load_yellow_taxi_data.py

这将：

• 从纽约市TLC公共数据门户下载黄色出租车行程数据（2024年1月至6月）
• 将文件上传到配置的GCS存储桶
• 验证上传成功

配置

编辑 load_yellow_taxi_data.py 自定义：

• BUCKET_NAME -目标地面军事系统铲斗
• MONTHS -下载月份列表
• DOWNLOAD_DIR -本地下载目录
• CHUNK_SIZE -上传块大小

Bigquery-玩SQL

-- Start with creating external table from Parquet files:
-- Creating external table referring to gcs path
CREATE OR REPLACE EXTERNAL TABLE `evident-axle-339820.nytaxi.external_yellow_tripdata`
OPTIONS (
  format = 'parquet',
  uris = ['gs://evident-axle-339820-hw3-2025/yellow_tripdata_2024-*.parquet']
)
;

-- Q1 What is count of records for the 2024 Yellow Taxi Data?
SELECT  COUNT(*)
FROM `evident-axle-339820.nytaxi.external_yellow_tripdata`

-- answer - 20_332_093 - for the first half of 2024
;

-- Create a non partitioned table from external table
CREATE OR REPLACE TABLE evident-axle-339820.nytaxi.yellow_tripdata_non_partitioned AS
SELECT * FROM evident-axle-339820.nytaxi.external_yellow_tripdata
;

-- Q2 What is the estimated amount of data that will be read when this query is executed on the External Table and the Table?
SELECT  COUNT (DISTINCT PULocationID)
-- FROM `evident-axle-339820.nytaxi.external_yellow_tripdata`
FROM evident-axle-339820.nytaxi.yellow_tripdata_non_partitioned

-- answer - 262 PULocationIDs, 0B from external and 155.12 MB when run on non-partitioned materialised table
;

-- Q3
SELECT  COUNT (DISTINCT PULocationID) AS count_pickup_location_id,
        COUNT (DISTINCT DOLocationID) AS count_dropoff_location_id
FROM evident-axle-339820.nytaxi.yellow_tripdata_non_partitioned

-- answer - query will process 310.24 MB when run - from 2 columns
;

-- Q4 How many records have a fare_amount of 0?
SELECT  COUNT(*) 
FROM evident-axle-339820.nytaxi.yellow_tripdata_non_partitioned
WHERE fare_amount = 0

-- answer - 8333
;

-- Create a partitioned table from external table
CREATE OR REPLACE TABLE evident-axle-339820.nytaxi.yellow_tripdata_partitioned
PARTITION BY
  DATE(tpep_pickup_datetime) AS
SELECT * FROM evident-axle-339820.nytaxi.yellow_tripdata_non_partitioned
;

-- Impact of partition - we have data only for 2024 but unpartitioned 
-- table does not know it in advance, so it has to run a query.
-- This query will process 310.24 MB when run.
SELECT DISTINCT(VendorID)
FROM evident-axle-339820.nytaxi.yellow_tripdata_non_partitioned
WHERE DATE(tpep_pickup_datetime) BETWEEN '2019-06-01' AND '2019-06-30'
;

-- This query will process 0 B when run:
-- because we have data only for 2024, nothing for 2019 - this is why 0 B .
SELECT DISTINCT(VendorID)
FROM evident-axle-339820.nytaxi.yellow_tripdata_partitioned
WHERE DATE(tpep_pickup_datetime) BETWEEN '2019-06-01' AND '2019-06-30'
;

-- Q6 - query to retrieve the distinct VendorIDs between tpep_dropoff_datetime 2024-03-01 and 2024-03-15 (inclusive). 
SELECT DISTINCT(VendorID)
FROM evident-axle-339820.nytaxi.yellow_tripdata_non_partitioned
-- FROM evident-axle-339820.nytaxi.yellow_tripdata_partitioned
WHERE DATE(tpep_pickup_datetime) BETWEEN '2024-03-01' AND '2024-03-15'
-- non-part -  process 310.24 MB when run
-- part - will process 26.85 MB when run.
;

-- Q9. Write a `SELECT count(*)` query FROM the materialized table you created. 
-- How many bytes does it estimate will be read? Why?
SELECT COUNT(*)
FROM evident-axle-339820.nytaxi.yellow_tripdata_non_partitioned
-- This query will process 0 B when run. 
-- Look at table info - Number of rows 20,332,093
;

本地DBT和DuckDB设置

本指南将引导您使用DuckDB和dbt设置本地分析工程环境。

 

\[！注意\] *本指南将解释如何手动进行设置。如果你想要一个额外的挑战，请尝试使用Docker Compose或Python虚拟环境运行此设置。*

重要：所有dbt命令都必须从内部运行 taxi_rides_ny/ 目录。以下设置步骤将指导您完成：

1. 安装必要的工具
2. 配置与DuckDB的连接
3. 加载纽约市出租车数据
4. 验证一切是否正常

步骤1：安装DuckDB

DuckDB是一个快速、进程内的SQL数据库，非常适合本地分析工作负载。要安装DuckDB，请按照 官方网站 针对您的特定操作系统。

\[！提示\] *您可以通过两种方式安装DuckDB。您可以安装CLI或为您喜欢的编程语言安装客户端API（对于Python，您可以使用 pip install duckdb 或更好 uv add duckdb).*

步骤2：安装dbt

pip install dbt-duckdb # uv add dbt-duckdb

这将安装：

• dbt-core：核心dbt框架
• dbt-duckdb：用于dbt的DuckDB适配器

步骤3：配置dbt配置文件

由于此存储库已包含dbt项目(taxi_rides_ny/)，你不需要跑 dbt init。相反，您需要配置dbt配置文件以连接到DuckDB。 复制 https://github.com/DataTalksClub/data-engineering-zoomcamp/tree/main/04-analytics-engineering/taxi_rides_ny 从github作为我使用的zip文件 https://download-directory.github.io/ 网站-只需复制/粘贴文件夹url，稍后解压缩目录即可。

创建或更新 ~/.dbt/profiles.yml

dbt概要文件告诉dbt如何连接到数据库。创建或更新文件 ~/.dbt/profiles.yml 内容如下：

taxi_rides_ny:
  target: dev
  outputs:
    # DuckDB Development profile
    dev:
      type: duckdb
      path: taxi_rides_ny.duckdb
      schema: dev
      threads: 1
      extensions:
        - parquet
      settings:
        memory_limit: '8GB'
        preserve_insertion_order: false

    # DuckDB Production profile
    prod:
      type: duckdb
      path: taxi_rides_ny.duckdb
      schema: prod
      threads: 1
      extensions:
        - parquet
      settings:
        memory_limit: '8GB'
        preserve_insertion_order: false

# Troubleshooting:
# - If you have less than 4GB RAM, try setting memory_limit to '1GB'
# - If you have 16GB+ RAM, you can increase to '4GB' for faster builds
# - Expected build time: 5-10 minutes on most systems

步骤4：下载并摄取数据

现在您的dbt配置文件已经配置好，让我们将出租车数据加载到DuckDB中。导航到dbt项目目录并运行摄入脚本

import duckdb
import requests
from pathlib import Path

BASE_URL = "https://github.com/DataTalksClub/nyc-tlc-data/releases/download"

def download_and_convert_files(taxi_type):
    data_dir = Path("data") / taxi_type
    data_dir.mkdir(exist_ok=True, parents=True)

    for year in [2019, 2020]:
        for month in range(1, 13):
            parquet_filename = f"{taxi_type}_tripdata_{year}-{month:02d}.parquet"
            parquet_filepath = data_dir / parquet_filename

            if parquet_filepath.exists():
                print(f"Skipping {parquet_filename} (already exists)")
                continue

            # Download CSV.gz file
            csv_gz_filename = f"{taxi_type}_tripdata_{year}-{month:02d}.csv.gz"
            csv_gz_filepath = data_dir / csv_gz_filename

            response = requests.get(f"{BASE_URL}/{taxi_type}/{csv_gz_filename}", stream=True)
            response.raise_for_status()

            with open(csv_gz_filepath, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    f.write(chunk)

            print(f"Converting {csv_gz_filename} to Parquet...")
            con = duckdb.connect()
            con.execute(f"""
                COPY (SELECT * FROM read_csv_auto('{csv_gz_filepath}'))
                TO '{parquet_filepath}' (FORMAT PARQUET)
            """)
            con.close()

            # Remove the CSV.gz file to save space
            csv_gz_filepath.unlink()
            print(f"Completed {parquet_filename}")

def update_gitignore():
    gitignore_path = Path(".gitignore")

    # Read existing content or start with empty string
    content = gitignore_path.read_text() if gitignore_path.exists() else ""

    # Add data/ if not already present
    if 'data/' not in content:
        with open(gitignore_path, 'a') as f:
            f.write('\n# Data directory\ndata/\n' if content else '# Data directory\ndata/\n')

if __name__ == "__main__":
    # Update .gitignore to exclude data directory
    update_gitignore()

    for taxi_type in ["yellow", "green"]:
        download_and_convert_files(taxi_type)

    con = duckdb.connect("taxi_rides_ny.duckdb")
    con.execute("CREATE SCHEMA IF NOT EXISTS prod")

    for taxi_type in ["yellow", "green"]:
        con.execute(f"""
            CREATE OR REPLACE TABLE prod.{taxi_type}_tripdata AS
            SELECT * FROM read_parquet('data/{taxi_type}/*.parquet', union_by_name=true)
        """)

    con.close()

此脚本下载2019-2020年的黄色和绿色出租车数据，创建 prod 并将原始数据加载到DuckDB中。下载可能需要几分钟，具体取决于您的互联网连接。

！！！注意！！！结果 taxi_rides_ny\taxi_rides_ny.duckdb 文件大小约为3GB-将.duckdb添加到gitignore-不要将其提交到github。..

步骤5：测试dbt连接

验证dbt是否可以连接到DuckDB数据库：

uv run dbt debug

步骤6：安装dbt高级用户扩展（VS代码用户）

如果您使用的是Visual Studio代码，请安装 dbt高级用户 扩展以增强您的dbt开发体验。

什么是dbt高级用户？

dbt高级用户是一个VS代码扩展，它提供：

• dbt模型的SQL语法高亮显示和格式化
• 内联列级沿袭可视化
• dbt模型、源代码和宏的自动完成
• 交互式文档预览
• 直接从编辑器编译和执行模型

为什么不使用官方dbt扩展？

dbt Labs发布了一个名为 dbt扩展 由新型dbt Fusion发动机提供动力。然而，这种扩展 需要dbt Fusion 并且不支持dbt-Core。

既然我们正在使用 dbt核心 通过DuckDB进行本地开发，我们需要维护社区 dbt高级用户AltimaeAI 取而代之的是扩展。此扩展：

• 与dbt Core（而不仅仅是dbt Cloud）无缝协作
• 支持所有dbt适配器，包括DuckDB
• 积极维护和开源
• 为本地开发提供丰富的功能集

安装

1. 打开VS代码
2. 转到扩展（Ctrl+Shift+X/Cmd+Shift+X）
3. 搜索“dbt高级用户”
4. 安装 dbt高级用户AltimaeAI （不是dbt Labs版本）

或者，从以下位置安装 VS代码市场.

\[！注意\] 此时，您的本地dbt环境已完全配置并准备好使用。

其他资源

• DuckDB文档
• dbt文档
• dbt-duckdb适配器
• 纽约市出租车数据字典

步骤7：运行duckdb UI，查看数据是否正确摄入

要从uv环境启动DuckDB UI并验证摄入的数据：

# From the project root, run:
uv run duckdb -ui taxi_rides_ny/taxi_rides_ny.duckdb

这将：

1. 在uv环境中使用CLI打开DuckDB
2. 在默认浏览器中启动内置UI
3. 连接到 taxi_rides_ny.duckdb 数据库文件

打开UI后，您可以运行SQL查询来验证数据，例如：

-- Check tables in the database
SHOW TABLES;

-- Count records in yellow_tripdata
SELECT COUNT(*) FROM prod.yellow_tripdata;

-- List schema
SELECT * FROM information_schema.tables;

或者运行 uv run verify_data.py 查看数据是否已加载。

步骤8：检查当前配置

第一张cd指向DBT文件所在的目录：

cd taxi_rides_ny

然后尝试从那里运行测试构建：

uv run dbt build

理想情况下，它应该没有错误地完成——如果有的话，请读取构建错误。.. 如果没有错误，请尝试检查prod构建：

uv run dbt run --select prod

有用的DBT命令是（列表）： （最好先从venv和cd taxi_rides_ny中选择一个Python解释器）

# 0. INIT - run once! - builds dbt project
uv run dbt init

# 1. DEBUG - checks database connection
uv run dbt debug

# 2. SEED - ingest, uploads or materialises seeds - in our case csv files for zones and ppayment types
uv run dbt seed 

# 3. RUN - less heavy than build - tries to compile models and materialise it
uv run dbt run

# 4. BUILD - heavy - builds all models + runs tests + materialises seeds.... etc etc
uv run dbt build 

# 5. DEPS - short for dependencies - installs packages
uv run dbt deps 

# 6. COMPILE - compiles sql files without jinja - ready to send to actual database engine - into target / compile folder - if I need to spot jinja errors...
uv run dbt compile 

# 7. TEST - run all tests from tests folder
uv run dbt test 

# 8. RETRY - starts from the point where last build failed.
uv run dbt retry

步骤9：如果一切正常-查询duckdb UI：

在测试了dbt-build之后，您可以构建prod模型并将其具体化到duckdb数据库中，这将在duchdb数据库中创建最终的表。对我来说，它需要将RAM限制增加到32GB，或者它给出了内存溢出的错误。 请耐心等待——在平均i5 PC上构建所有表格可能需要30分钟左右。

uv run dbt build --select prod

启动duckdb UI开始测试：

cd taxi_rides_ny
uv run duckdb -ui taxi_rides_ny.duckdb

然后，您可以运行SQL查询来验证数据，例如：

-- Count of records in fct_monthly_zone_revenue?
-- 12184
SELECT COUNT(*)

from taxi_rides_ny.prod.fct_monthly_zone_revenue

或

-- Zone with highest revenue for Green taxis in 2020? 
-- East Harlem North
SELECT revenue_monthly_total_amount, pickup_zone
FROM taxi_rides_ny.prod.fct_monthly_zone_revenue
WHERE service_type = 'Green' 
  and year(revenue_month) = 2020
ORDER BY 1 DESC
LIMIT 10

或

-- Zone with highest revenue for Green taxis in 2020? 
-- East Harlem North
SELECT revenue_monthly_total_amount, pickup_zone
FROM taxi_rides_ny.prod.fct_monthly_zone_revenue
WHERE service_type = 'Green' 
  and year(revenue_month) = 2020
ORDER BY 1 DESC
LIMIT 10

或

-- Total trips for Green taxis in October 2019?
-- 384624
SELECT COUNT(*)
from taxi_rides_ny.prod.fct_trips
WHERE service_type = 'Green' 
  and month(pickup_datetime) = 10
  and year(pickup_datetime) = 2019

启动Bruin-简单管道

此管道是Bruin项目的一个简单示例。它演示了如何使用 bruin CLI用于构建和运行管道。 DuckDB因其简洁性而被选中。此设置假定DuckDB可用；你可以交换 duckdb.sql 资产类型。

管道包括以下示例资产：

• dataset.players：一个将国际象棋选手数据加载到DuckDB中的ingestr资产。
• dataset.player_stats：从以下内容构建表的DuckDB SQL资产 dataset.players.
• my_python_asset：打印消息的Python资源。

设置

此模板包括 .bruin.yml 带有示例DuckDB和国际象棋连接。您可以根据需要替换或扩展您的连接和环境。

这是一个样品 .bruin.yml 文件：

default_environment: default
environments:
  default:
    connections:
      duckdb:
        - name: "duckdb-default"
          path: "duckdb.db"
      chess:
        - name: "chess-default"
          players:
            - "MagnusCarlsen"
            - "Hikaru"

您只需使用以下命令即可切换环境 --environment 标志，例如：

bruin validate --environment production . 

运行管道

bruin CLI可以运行整个管道或任何下游任务：

bruin run .

Starting the pipeline execution...

[18:42:58] Running:  my_python_asset
[18:42:58] Running:  dataset.players
[18:42:58] [my_python_asset] >> warning: `--no-sync` has no effect when used outside of a project
[18:42:58] [my_python_asset] >> hello world
[18:42:58] Finished: my_python_asset (191ms)
⋮
[18:43:04] Finished: dataset.player_stats:player_count:not_null (24ms)
[18:43:04] Finished: dataset.player_stats:player_count:positive (33ms)
[18:43:04] Finished: dataset.player_stats:name:unique (42ms)

==================================================

PASS my_python_asset 
PASS dataset.players 
PASS dataset.player_stats .....

bruin run completed successfully in 5.439s

 ✓ Assets executed      3 succeeded
 ✓ Quality checks       5 succeeded

您还可以运行单个任务：

bruin run assets/my_python_asset.py                         

Starting the pipeline execution...

[23:00:02] Running:  my_python_asset
[23:00:02] >> warning: `--no-sync` has no effect when used outside of a project
[23:00:02] >> hello world
[23:00:02] Finished: my_python_asset (162ms)

==================================================

PASS my_python_asset 

bruin run completed successfully in 162ms

 ✓ Assets executed      1 succeeded

您可以选择传递 --downstream 标记以运行任务及其所有下游。

就是这样，你们都准备好了。幸福大厦！

如果你想深入挖掘，就跳进 概念 了解更多关于Bruin用于数据管道的基本概念。