如何在DBT（Data Build Tool）中跳过Seeds的表头与表尾

方法一：通过DBT Model过滤已加载的Seed数据 #

适合不想修改原始CSV的场景，先全量加载CSV到seed表，再通过model筛选有效行：

• 1. 正常配置并运行seed，将CSV加载到数据库：
     在dbt_project.yml中配置seed路径：

  seeds:
    your_project_name:
      raw_data:
        +schema: raw
  

  执行命令：dbt seed

• 2. 创建model过滤表头表尾：
     如果有可识别有效行的字段（比如表头行id为字符串ID，表尾行含Total关键字），SQL示例：

  {{ config(materialized='table') }}
  
  select *
  from {{ ref('raw_data') }}
  where id != 'ID'
    and not contains(description, 'Total')
  

  无标识字段时，用行号过滤（不同数据库语法略有差异）：

  • PostgreSQL/Redshift：

    select *
    from (
        select *, row_number() over () as row_num
        from {{ ref('raw_data') }}
    ) t
    where row_num > 1 -- 跳过第1行表头
      and row_num <= (select count(*) from {{ ref('raw_data') }}) - 2 -- 跳过最后2行表尾
    

  • BigQuery/Snowflake：

    select *
    from (
        select *, row_number() over (order by null) as row_num
        from {{ ref('raw_data') }}
    ) t
    where row_num > 1
      and row_num <= (select count(*) from {{ ref('raw_data') }}) - 2
    

方法二：预处理CSV文件（加载前清理） #

直接在加载前去掉表头表尾，让DBT加载干净的文件：

• 1. 编写Python脚本（clean_csv.py）处理原始CSV：

  import csv
  
  input_file = 'seeds/raw_data.csv'
  output_file = 'seeds/clean_raw_data.csv'
  skip_header = 1  # 跳过前1行表头
  skip_footer = 2  # 跳过最后2行表尾
  
  with open(input_file, 'r', newline='') as infile, open(output_file, 'w', newline='') as outfile:
      reader = csv.reader(infile)
      writer = csv.writer(outfile)
      
      # 跳过表头
      for _ in range(skip_header):
          next(reader)
      
      # 移除表尾行
      rows = list(reader)
      clean_rows = rows[:-skip_footer] if skip_footer > 0 else rows
      
      writer.writerows(clean_rows)
  

• 2. 先执行脚本再运行seed：

  python clean_csv.py && dbt seed --select clean_raw_data
  

• 3. 配置DBT加载处理后的CSV：

  seeds:
    your_project_name:
      clean_raw_data:
        +schema: raw
  

方法三：结合数据库文件格式跳过表头（表尾仍需过滤） #

部分数据库（如Snowflake）支持通过文件格式参数跳过表头，表尾仍需后续SQL过滤：

• 1. 编写宏创建自定义文件格式：
     在macros/create_file_format.sql中添加：

  {% macro create_custom_csv_format() %}
  create or replace file format custom_csv_format
  type = csv
  field_delimiter = ','
  skip_header = 1  -- 跳过1行表头
  {% endmacro %}
  

• 2. 配置seed使用该文件格式：

  seeds:
    your_project_name:
      raw_data:
        +schema: raw
        +file_format: custom_csv_format
  

• 3. 运行宏并加载seed：

  dbt run-operation create_custom_csv_format && dbt seed
  

  之后用方法一中的model逻辑过滤表尾行即可。

内容的提问来源于stack exchange，提问作者Hitesh Bhatt