3. 数据源接入与source配置

在 dbt 项目中，source 是连接“外部真实数据”与“dbt 内部模型”的第一道契约。它把散落在数据库里的原始表登记为有名字、有描述、有测试、有新鲜度监控的正式资产。没有 source 配置，模型里只能写 {{ ref('...') }} 或裸表名，血缘断裂、测试缺失、上游变更无人知晓。本节围绕 source 定义、freshness 检查、源数据测试、原始层建模策略与真实业务源表接入展开。

一、source 定义：把外部表登记为可治理资产

source 定义写在 YAML 文件中，通常放在 models/staging/ 或专门的 models/sources/ 目录下。一个 source 由 name（源系统名）、database、schema 和若干 tables 组成。例如电商业务中，订单库、用户库、支付库分别对应三个 source：

version: 2

sources:
  - name: ecommerce_order
    description: "订单交易库，MySQL 同步至数仓 ods 层"
    database: analytics
    schema: ods_order
    tables:
      - name: orders
        description: "订单主表，一行一个订单"
        columns:
          - name: order_id
            description: "订单唯一标识"
          - name: user_id
            description: "下单用户 ID"
          - name: order_status
            description: "订单状态：created/paid/shipped/completed/cancelled"
          - name: created_at
            description: "订单创建时间"
      - name: order_items
        description: "订单明细表，一行一个商品行"

定义完成后，模型里用 {{ source('ecommerce_order', 'orders') }} 引用。这样做有三个直接收益：一是血缘图中能看到上游源表节点；二是可以对源表列做测试；三是 dbt source freshness 能监控数据是否按时到达。

二、freshness 检查：监控数据是否“新鲜”

freshness 用于回答一个问题：源表最近一次数据更新距今多久？如果超过阈值，说明上游同步可能失败或延迟。配置写在 source 或 table 级别：

    freshness:
      warn_after: {count: 6, period: hour}
      error_after: {count: 12, period: hour}
    loaded_at_field: created_at

loaded_at_field 是判断新鲜度的时间列，通常是 created_at 或 updated_at。执行 dbt source freshness 后，dbt 会计算 max(loaded_at_field) 与当前时间的差值，输出 pass/warn/error 三态。示例结果：某订单表最新记录时间为 2024-06-01 08:15，当前时间 12:30，差值为 4 小时 15 分，低于 6 小时 warn 阈值，状态为 pass。若差值达到 8 小时，则触发 warn；超过 12 小时触发 error，通常应阻断下游任务。

三、源数据测试：在入口处拦截脏数据

源数据测试直接写在 source 的 columns 下，常用测试包括 not_null、unique、accepted_values 和 relationships。例如：

        columns:
          - name: order_id
            tests:
              - not_null
              - unique
          - name: order_status
            tests:
              - accepted_values:
                  values: ['created','paid','shipped','completed','cancelled']
          - name: user_id
            tests:
              - not_null

这些测试在 dbt test 时执行，失败会明确指向源表。实践中建议对源表只做“轻量但关键”的测试：主键唯一非空、状态枚举合法、外键存在。不要在源表上做复杂业务规则测试，那属于 staging 或中间层的职责。

四、原始层建模策略：staging 只做清洗与标准化

原始层（staging）模型通常以 stg_ 开头，一对一映射源表，只做四类操作：字段重命名、类型转换、简单清洗、派生基础字段。不做聚合、不做多表 join、不做业务逻辑。例如：

-- models/staging/stg_orders.sql
select
    order_id,
    user_id,
    cast(created_at as timestamp) as created_at,
    lower(trim(order_status)) as order_status,
    case when order_status = 'completed' then true else false end as is_completed
from {{ source('ecommerce_order', 'orders') }}

这样做的价值是：下游模型只依赖稳定的 staging 字段，源表改名或换库时只需改 source 定义和 staging 模型，不必全项目搜索替换。

五、真实业务源表接入的完整链路

把上述环节串起来，一次真实接入的流程是：确认源表结构与时间列 → 编写 source YAML → 配置 freshness → 添加关键测试 → 编写 staging 模型 → 执行 dbt source freshness 与 dbt test → 观察结果并决定是否放行下游。下图展示了这一链路中源表、source 定义、freshness、测试与 staging 模型之间的关系。

接入真实业务源表时，最容易忽略的是时间列语义。例如 created_at 表示业务发生时间，updated_at 表示记录最后修改时间，etl_loaded_at 表示数据入仓时间。freshness 应优先使用 etl_loaded_at 或 updated_at，因为 created_at 可能因补历史数据而失真。若源表没有可靠时间列，可退化为“行数变化监控”或“分区存在性检查”，但这属于补充手段，不能替代 freshness。

最后，建议把 source 接入纳入 CI：每次合并前执行 dbt source freshness 和 dbt test --select source:*，确保新增或修改的源表定义不会破坏既有链路。这样，source 才真正成为数据质量的第一道闸门，而不是一份静态文档。