The Medallion Data Architecture: An Introduction

retail-medallion/
├── data/
│   └── incoming/
│       └── orders_2026-07-19.csv    <= 由你手动创建
├── pipeline.py                      <= 由你手动创建
└── warehouse.duckdb                 <= 该数据库文件由管道创建

D:\projects\retail-medallion> python3 -m venv .venv
# Windows PowerShell: .\.venv\Scripts\Activate.ps1
# macOS/Linux: source .venv/bin/activate
D:\projects\retail-medallion> python3 -m pip install duckdb pytz tabulate

order_id,ordered_at,customer_id,region,amount,currency,status
1001,2026-07-19T09:10:00Z,C001,North,125.50,GBP,paid
1002,2026-07-19T10:05:00Z,C002,South,89.99,GBP,paid
1002,2026-07-19T10:05:00Z,C002,South,89.99,GBP,paid
1003,not-a-date,C003,North,45.00,GBP,paid
1004,2026-07-19T11:42:00Z,C004,West,-10.00,GBP,paid
1005,2026-07-19T12:20:00Z,C005,North,210.00,GBP,refunded

from __future__ import annotations

import hashlib
import sys
from pathlib import Path

import duckdb

DATABASE = Path("warehouse.duckdb")

def file_hash(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as source:
        for block in iter(lambda: source.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()

def initialise(connection: duckdb.DuckDBPyConnection) -> None:
    connection.execute("CREATE SCHEMA IF NOT EXISTS bronze")
    connection.execute("CREATE SCHEMA IF NOT EXISTS silver")
    connection.execute("CREATE SCHEMA IF NOT EXISTS gold")
    connection.execute("""
        CREATE TABLE IF NOT EXISTS bronze.ingestion_batches (
            source_hash VARCHAR PRIMARY KEY,
            source_file VARCHAR NOT NULL,
            ingested_at TIMESTAMPTZ NOT NULL DEFAULT current_timestamp
        )
    """)
    connection.execute("""
        CREATE TABLE IF NOT EXISTS bronze.orders_raw (
            order_id VARCHAR,
            ordered_at VARCHAR,
            customer_id VARCHAR,
            region VARCHAR,
            amount VARCHAR,
            currency VARCHAR,
            status VARCHAR,
            source_file VARCHAR NOT NULL,
            source_hash VARCHAR NOT NULL,
            ingested_at TIMESTAMPTZ NOT NULL
        )
    """)

def ingest_bronze(connection: duckdb.DuckDBPyConnection, source: Path) -> bool:
    source = source.resolve()
    digest = file_hash(source)
    already_loaded = connection.execute(
        "SELECT 1 FROM bronze.ingestion_batches WHERE source_hash = ?", [digest]
    ).fetchone()
    if already_loaded:
        print(f"Skipping {source.name}: this exact file has already been loaded")
        return False

    connection.begin()
    try:
        connection.execute(
            """
            INSERT INTO bronze.orders_raw
            SELECT
                order_id, ordered_at, customer_id, region, amount,
                currency, status, ?, ?, current_timestamp
            FROM read_csv(?, header = true, all_varchar = true)
            """,
            [source.name, digest, str(source)],
        )
        connection.execute(
            """INSERT INTO bronze.ingestion_batches (source_hash, source_file)
            VALUES (?, ?)""",
            [digest, source.name],
        )
        connection.commit()
    except Exception:
        connection.rollback()
        raise
    print(f"Loaded {source.name} into bronze")
    return True

def build_silver(connection: duckdb.DuckDBPyConnection) -> None:
    connection.execute("""
        CREATE OR REPLACE TEMP VIEW typed_orders AS
        SELECT
            trim(order_id) AS order_id,
            try_cast(ordered_at AS TIMESTAMPTZ) AS ordered_at,
            trim(customer_id) AS customer_id,
            upper(trim(region)) AS region,
            try_cast(amount AS DECIMAL(18, 2)) AS amount,
            upper(trim(currency)) AS currency,
            lower(trim(status)) AS status,
            source_file,
            source_hash,
            ingested_at,
            row_number() OVER (
                PARTITION BY trim(order_id)
                ORDER BY ingested_at DESC, source_file DESC
            ) AS duplicate_rank
        FROM bronze.orders_raw
    """)
    valid = """
        order_id IS NOT NULL AND order_id <> ''
        AND ordered_at IS NOT NULL
        AND customer_id IS NOT NULL AND customer_id <> ''
        AND amount IS NOT NULL AND amount >= 0
        AND currency IN ('GBP', 'EUR', 'USD')
        AND status IN ('paid', 'refunded', 'cancelled')
        AND duplicate_rank = 1
    """
    connection.execute(f"""
        CREATE OR REPLACE TABLE silver.orders AS
        SELECT * EXCLUDE (duplicate_rank)
        FROM typed_orders
        WHERE {valid}
    """)
    connection.execute(f"""
        CREATE OR REPLACE TABLE silver.orders_quarantine AS
        SELECT
            * EXCLUDE (duplicate_rank),
            CASE
                WHEN duplicate_rank > 1 THEN 'duplicate order_id'
                WHEN ordered_at IS NULL THEN 'invalid ordered_at'
                WHEN amount IS NULL THEN 'invalid amount'
                WHEN amount < 0 THEN 'negative amount'
                WHEN currency NOT IN ('GBP', 'EUR', 'USD') THEN 'unsupported currency'
                WHEN status NOT IN ('paid', 'refunded', 'cancelled') THEN 'invalid status'
                ELSE 'missing required value'
            END AS rejection_reason
        FROM typed_orders
        WHERE NOT ({valid})
    """)

def build_gold(connection: duckdb.DuckDBPyConnection) -> None:
    connection.execute("""
        CREATE OR REPLACE TABLE gold.daily_sales_by_region AS
        SELECT
            cast(ordered_at AS DATE) AS order_date,
            region,
            currency,
            count(*) FILTER (WHERE status = 'paid') AS paid_orders,
            sum(amount) FILTER (WHERE status = 'paid') AS gross_sales,
            count(*) FILTER (WHERE status = 'refunded') AS refunded_orders,
            sum(amount) FILTER (WHERE status = 'refunded') AS refunded_value
        FROM silver.orders
        GROUP BY order_date, region, currency
        ORDER BY order_date, region, currency
    """)

def check_quality(connection: duckdb.DuckDBPyConnection) -> None:
    duplicate_count = connection.execute(
        "SELECT count(*) - count(DISTINCT order_id) FROM silver.orders"
    ).fetchone()[0]
    null_key_count = connection.execute(
        "SELECT count(*) FROM silver.orders WHERE order_id IS NULL"
    ).fetchone()[0]
    if duplicate_count or null_key_count:
        raise RuntimeError("Silver quality contract failed")

def print_query(connection: duckdb.DuckDBPyConnection, query: str) -> None:
    result = connection.execute(query)
    print(" | ".join(column[0] for column in result.description))
    for row in result.fetchall():
        print(" | ".join("NULL" if value is None else str(value) for value in row))

def main(source: Path) -> None:
    with duckdb.connect(str(DATABASE)) as connection:
        initialise(connection)
        ingest_bronze(connection, source)
        build_silver(connection)
        check_quality(connection)
        build_gold(connection)
        print("\nGold output")
        print_query(connection, "SELECT * FROM gold.daily_sales_by_region")
        print("\nQuarantined records")
        print_query(
            connection,
            """SELECT order_id, ordered_at, amount, rejection_reason
            FROM silver.orders_quarantine""",
        )

if __name__ == "__main__":
    if len(sys.argv) != 2:
        raise SystemExit("Usage: python pipeline.py path/to/orders.csv")
    main(Path(sys.argv[1]))

使用以下命令运行

python3 pipeline.py data/incoming/orders_2026-07-19.csv

已将 orders_2026-07-19.csv 加载到青铜层

Gold output
order_date | region | currency | paid_orders | gross_sales | refunded_orders | refunded_value
2026-07-19 | NORTH  | GBP      | 1           | 125.50      | 1               | 210.00
2026-07-19 | SOUTH  | GBP      | 1           | 89.99       | 0               | NULL

Quarantined records
order_id | ordered_at                | amount | rejection_reason
1004     | 2026-07-19 12:42:00+01:00 | -10.00 | negative amount
1003     | NULL                      | 45.00  | invalid ordered_at
1002     | 2026-07-19 11:05:00+01:00 | 89.99  | duplicate order_id

import duckdb
from tabulate import tabulate

def show_table(
    connection: duckdb.DuckDBPyConnection,

    query: str,
) -> None:
    result = connection.execute(query)
    headers = [column[0] for column in result.description]

    print(f"\n{title}")
    print(tabulate(result.fetchall(), headers=headers, tablefmt="psql"))

with duckdb.connect("warehouse.duckdb") as connection:
    show_table(
        connection,
        "BRONZE - Raw orders",
        """
        SELECT
            order_id,
            ordered_at,
            customer_id,
            region,
            amount,
            currency,
            status,
            source_file
        FROM bronze.orders_raw
        ORDER BY order_id
        """,
    )

    show_table(
        connection,
        "SILVER - Validated orders",
        """
        SELECT
            order_id,
            ordered_at,
            customer_id,
            region,
            amount,
            currency,
            status
        FROM silver.orders
        ORDER BY order_id
        """,
    )

Which results in the following output.

## 总结

作为数据库和数据工程师，我们经常在ETL作业中听到关于勋章模式（Medallion pattern）的讨论，老实说，你可能已经多次实现了它的简化版本。在本文中，我试图从基本原理出发，为你展示如何实现一个实用的勋章架构。

不要误解我。我给你展示的例子确实是一个玩具示例，它使用了有限的输入数据和本地数据库，但构建更大规模生产系统的必要原则已经具备。

在生产环境中，你必须决定是使用企业级关系型数据库管理系统（如Postgres或Oracle），还是使用基于云的对象存储（如AWS S3）。如果选择后者，你需要考虑使用哪种事务表存储格式（如Hudi、Delta表或Iceberg）。你还需要考虑是否需要使用管道编排工具（如Airflow或Dagster）。

我甚至还没有谈到分层边界所需的自动化检查类型。例如包括：

- 青铜层行数统计和数据源完整性
- 银层键唯一性、接受值检查和引用完整性
- 金层与银层总量的对账
- 新鲜度和数据量阈值
- 隔离率和模式漂移的预警

但这些只是锦上添花。关键是要理解勋章模式的基本原理，并认识到它如何以及在哪些场景下可以融入你现有的或新的ETL管道中。

勋章架构之所以有效，是因为它使数据中的差异变得可见。接收到的数据与验证后的数据不同，而验证后的数据并不自动适用于特定的业务决策。

作者：Thomas Reid

查看Thomas Reid的所有文章

数据建模

,

ETL管道

勋章架构

编程

Python

分享本文

- 在Facebook上分享
- 在LinkedIn上分享
- 在X上分享

Towards Data Science 是一份社区出版物。提交你的见解以触达全球受众，并通过TDS作者支付计划获得收益。

将href更新为你的实际投稿链接

为TDS撰写文章

✦ 结束CTA ✦