dbt Connector

步骤一：声明源

源在以sources:键嵌套的.yml文件中定义。models/<filename>.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。

步骤二：从源表中选择数据

声明源后，可通过{{ source() }}函数在模型中引用它：

models/orders.sql

SELECT
  ...
  FROM {{ source('jaffle_shop', 'orders') }}
  LEFT JOIN {{ source('jaffle_shop', 'customers') }} USING (customer_id)

dbt 会将其编译为完整的表名

SELECT
  ...
  FROM raw.jaffle_shop.orders
  LEFT JOIN raw.jaffle_shop.customers USING (customer_id)

使用{{ source() }}函数会创建模型与源表之间的依赖关系。模型 orders 与源表 jaffle_shop.customers、jaffle_shop.orders 的依赖关系图如下：

步骤三：测试与文档化源

• 为源添加数据测试。

• 为源添加描述，这些描述会作为文档站点的一部分生成。

模型中可配置属性models/<filename>.yml

version: 2

sources:
  - name: jaffle_shop
    description: This is a replica of the Postgres database used by our app
    tables:
      - name: orders
        description: >
          One record per order. Includes cancelled and deleted orders.
        columns:
          - name: id
            description: Primary key of the orders table
            tests:
              - unique
              - not_null
          - name: status
            description: Note that the status can change over time

      - name: ...

  - name: ...

步骤一：声明数据新鲜度

在源配置models/<filename>.yml中添加freshness和loaded_at_field：

version: 2

sources:  
   - name: jaffle_shop    
     database: raw    
     freshness: # 默认新鲜度阈值    
       warn_after: {count: 12, period: hour}      
       error_after: {count: 24, period: hour}    
     loaded_at_field: _etl_loaded_at    
     
     tables:      
       - name: orders        
         freshness: # 更严格的时间阈值      
           warn_after: {count: 6, period: hour}          
           error_after: {count: 12, period: hour}      
        
       - name: customers # 继承父级配置      
       
       
       - name: product_skus        
         freshness: null # 不检查此表的新鲜度

步骤二：验证数据新鲜度

运行dbt source freshness命令检测源数据的新鲜度：

$ dbt source freshness

dbt 会根据freshness配置生成查询：

SELECT
  MAX(_etl_loaded_at) AS max_loaded_at,
  CONVERT_TIMEZONE('UTC', CURRENT_TIMESTAMP()) AS calculated_at
FROM raw.jaffle_shop.orders

若检测到数据未达到新鲜度阈值，会触发警告或错误。