本章节介绍在dbt项目中，如何定义数据源模型。

定义并引入数据源

通过Extract和Load方式加载到仓库中的数据，可以使用dbt中的sources组件进行定义和描述。通过在dbt中将这些数据集(表)声明为sources，可以实现如下功能：

• 在dbt模型models中，通过使用{{ source }}函数来查询数据源的数据集(表)，这样就可以实现数据间的血缘关系。
• 可以对数据源进行测试验证。
• 能够计算数据源的数据抽取刷新情况。

dbt项目中的数据源定义在.yml文件中的sources关键字下。如下所示，在models目录中，数据源yml文件的内容为：

version: 2

sources:
  - name: jaffle_shop
    database: raw  
    schema: jaffle_shop  
    tables:
      - name: orders
      - name: customers

  - name: stripe
    tables:
      - name: payments

数据源配置文件中的 schema 通常与 name 相同，除非使用了不同于现有 schema 的数据源名称，此时需要添加 schema 关键字。

一旦数据源被定义完毕，在模型model中就可以通过使用source函数调用数据源。例如在models文件目录下定义一个order模型，文件名称为：models/order.sql

select
  ...

from {{ source('jaffle_shop', 'orders') }}

left join {{ source('jaffle_shop', 'customers') }} using (customer_id)

dbt将把它编译成完整的表名，编译后的文件在target/compiled/目录下:

select
  ...

from raw.jaffle_shop.orders

left join raw.jaffle_shop.customers using (customer_id)

在models的代码中使用{{ source() }}函数可以创建模型model和数据源间的依赖关系，并在文档中以图的形式显示出来：

创建完成数据源后，可以针对数据源中的数据集(表)进行测试，同时对数据源和数据集(表)添加注释后，dbt会将注释添加到 生成的web文档中。

version: 2

sources:
  - name: jaffle_shop
    description: This is a replica of the Postgres database used by our app
    tables:
      - name: orders
        description: >
          每条记录为一个订单. 包括取消和删除订单操作.
        columns:
          - name: id
            description: 订单表的主键
            tests:
              - unique # 添加对表中字段唯一测试
              - not_null # 添加对表中字段非空测试
          - name: status
            description: 超时后订单状态会改变

      - name: ...

  - name: ...

校验数据源的刷新

数据源通常是由数据同步工具定期加载到数据仓库中，供下层数仓模型进行消费。如果数据同步异常，可能会导致上层数据模型业务逻辑出现问题。dbt通过使用sources freshness属性配置选项，对数据源或表的限制，来验证数据源的加载是否符合同步预期，以确认数据处理是否处于正确状态。这对于数据处理的准确性非常有用。

要配置数据源刷新状态信息，在yml配置文件中添加一个freshness属性代码块，并在表声明中添加loadd_at_field属性关键字:

version: 2

sources:
  - name: jaffle_shop
    database: raw
    freshness: # 针对整个数据源级别的数据刷新情况
      warn_after: {count: 12, period: hour} # dbt source freshness命令执行时间与loaded_at_field属性定义的时间差值需要小于count属性值，否则执行freshness命令时报警
      error_after: {count: 24, period: hour}
# 同上定义
    loaded_at_field: _etl_loaded_at

    tables:
      - name: orders
        freshness: # 针对orders表定义数据刷新情况，表级别的定义
          warn_after: {count: 6, period: hour}
          error_after: {count: 12, period: hour}

      - name: customers # 如果表级别没有定义数据源的刷新，则默认数据源级别


      - name: product_skus
        freshness: null # do not check freshness for this table

在freshness属性代码块中，可以提供warn_after和error_after中的一个或两个告警级别。如果两者都没有提供，那么dbt将不会为这个数据源中的表计算刷新快照。

此外，还需要loadd_at_field关键子来计算表的刷新情况。如果没有提供loadd_at_field关键字，那么dbt将不会计算表的刷新情况。

这些配置存在层级关系，因此为一个数据源指定的刷新校验规则和和loadd_at_field值，将传递到该源下一层级定义的所有表中。如果一个源中的所有表都具有相同的loadd_at_field，只需要在上级源定义中指定一次即可。

使用dbt source freshness命令进行对数据源刷新情况进行校验。

$ dbt source freshness

在后台，dbt使用freshness属性来构造一个select语句查询，如下所示，您可以在查询执行日志中找到此执行语句。

select
  max(_etl_loaded_at) as max_loaded_at,
  convert_timezone('UTC', current_timestamp()) as snapshotted_at
from raw.jaffle_shop.orders

此查询的结果用于确定数据源源是否符合数据刷新规则：

sources数据源的yml配置文件中，通过使用freshness属性代码块、test属性代码块，实现了对数据源刷新频率的校验和数据质量的检查，避免由于数据异常引起后续的数据处理逻辑错误。因此，通过使用dbt工具对数据项目进行构建，很容易帮助数据开发人员提高工作效率，保证项目交付质量。