【dbt】数据底座建设案例文章目录前言一、DBT二、使用步骤2.1. 项目初始化2.1.1.首次初始化2.1.2.中断初始化2.1.3.完成初始化2.1.4.检查是否完成配置2.2.创建模型2.2.1.全量 跨shcema取数说明2.2.1.1.方法1不使用source直接使用SQL语句完成。2.2.1.2.方法2使用source需要配置source.yml文件2.2.1.3.全量2.2.2.增量 : 首次全量非首次增量2.2.3.视图 数据血缘2.2.4.Marco宏2.3. 数据管理2.3.1.生成数据文档2.3.2.编写yml文件2.3.3.数据质量2.3.3.1.通用规则2.3.3.2.自定义规则2.3.3.3.执行2.3.3.4.对比gx前言在大模型与数据分析领域数据底座建设十分重要。接到需求需要对数仓已有明细进行整合写入新的目标库作为AI应用与数据分析的支撑数据。使用dbt完成包括以下内容交付数据库中的数据、使用dbt进行数据更新、dbt数据血缘、dbt形成文档、统计口径文档。一、DBT介绍详见官网。官网安装教程二、使用步骤本章使用的是vscode及插件power user for dbt。不使用插件dbt是因为dbt的数据血缘需要在网页中呈现power user for dbt可以在编译器窗口中展示。完成插件安装后需要在设置中添加一个文件关联# settings.jsonfiles.associations:{*.sql:jinja-sql},2.1. 项目初始化2.1.1.首次初始化# 在终端执行dbt init需要根据提示输入相关信息一般如下项目名称会帮你建立一个项目文件夹数据库类型默认是posrgrehostportusernamepassworddbname数据库名称填写目标数据库。schema模式名称填写目标模式名称。threads默认12.1.2.中断初始化建议准备好再执行和填写。试过填错中断ctrlC重新初始化无法用相同的项目名称。解决方法删掉该文件下的dbt_project.yml文件把整个文件夹删掉再次执行dbt init重新初始化。2.1.3.完成初始化完成后打开文件夹可以看到如下结构analysesdbt_packageslogsmacros存放可复用的 Jinja 宏类似 SQL 函数。models存放主要的SQL数据模型逻辑。seeds存放小的静态 CSV 文件如码表、映射表dbt 可以直接将其加载进数据库。snapshots存放用于记录数据历史变化的“快照”代码。targetdbt 编译后生成的 SQL 文件和运行结果存放的目录。test存放数据质量测试脚本。dbt_project.ymlREADME.md打开dbt_project.yml可以看到相关的配置信息。核心逻辑配置全局、局部设置物化存储的格式包括table表、view视图、incremental增量表也可以在具体的SQL文件中单独覆盖该全局设置。具体如下# yml文件models:materialized:table# 全局所有模型都是表最低filename:# 你的项目名称example:# 项目下的文件夹materialized:view# 局部特例只有 example 文件夹下的是视图中# SQL文件# 在SQL文件中设置最高{{ config(materialized...) }}{{config( materialized incremental )}}二编新增在dbt_project.yml中添加文档参数可以在dbt数据更新中把models.yml的注释同步到表中。models:filename:example:materialized:viewtables:materialized:tablepersist_docs:relation:true# 同步表注释columns:true# 同步列注释正常情况下还需要有profiles.yml、user.yml文件。这俩文件自动生成在根目录dbt文件下参考路径如下C:\Users\用户名称\.dbt打开profiles.yml可以看见初始化时填写的信息记录在该文件中。关于该配置文件如有建立建议复制一份在项目根目录下方便后续文件打包与查阅.user.yml也建议复制一份。并且重新初始化会覆盖原有配置信息profiles.yml文件属性如下图如果没有生成PG可以参考如下格式新建profiles.yml文件参考自知乎其他数据库可查找官网profiles.yml的说明文档。default:target:devoutputs:dev:type:postgreshost:[hostname]user:[username]password:[password]port:[port]dbname:[database name]schema:[dbt schema]threads:[1 or more]keepalives_idle:0# default 0, indicating the system defaultsearch_path:[optional,override the default postgres search_path]role:[optional,set the role dbt assumes when executing queries]sslmode:[optional,set the sslmode used to connect to the database]2.1.4.检查是否完成配置在终端项目文件下执行以下命令# 执行命令dbt debug# 执行结果09:38:23 All checks passed!# 配置完成可以看到使用的配置文件路径使用的是根目录路径下的还是项目目录下的profiles.yml2.2.创建模型完成配置可以开始创建模型。我在models文件夹下分别创建文件夹tables、views不把创建模型放在example中。在models/tables文件夹下创建sql文件。2.2.1.全量 跨shcema取数说明注意1profiles.yml中只配置了目标数据库而没有要求配置来源数据库。是由于DBT默认数据已完成集成仅支持从Schema A迁移转化至Schema B。因此需要确保配置文件中的数据库为来源数据库并且也是目标数据库配置文件中的schema为目标shcema后续步骤为跨schema取数、转换、写入。注意2sql文件名就是表名。2.2.1.1.方法1不使用source直接使用SQL语句完成。-- t_dim_date.sql{{ config(materializedtable,description日期维度表)}}SELECT*FROMpublic.t_dim_date2.2.1.2.方法2使用source需要配置source.yml文件sourece.yml需要创建在models路径下。# models/sources.ymlversion:2sources:-name:publicschema:public# 确保这个 schema 在你连接的那个库里存在tables:-name:t_dim_date-- t_dim_date.sql{{ config(materializedtable,description日期维度表)}}SELECT*FROM{{ source(public,t_dim_date)}}2.2.1.3.全量不需要在代码里写TRUNCATE或DELETE。在 dbt 中只要物化方式是table每一次dbt run都是一次全量覆盖更新。你只需要在 SQL 中控制生成的日期范围例如每年年初把 SQL 里的结束日期往后改一年然后重新运行即可。# 运行整个dbt项目dbt run# 运行单个文件dbt run--modlesfilename# 例如dbt run--modelst_dim_date.sql# 成功返回内容07:22:251of1START sql table model dbt_models.t_dim_date.............................[RUN]07:22:251of1OK created sql table model dbt_models.t_dim_date........................[SELECT1096in0.41s]07:22:25 07:22:25 Finished running1table modelin0hours0minutes and1.15seconds(1.15s). 07:22:25 07:22:25 Completed successfully这时候去对应的数据库schema中查看可以查到该表。2.2.2.增量 : 首次全量非首次增量{{ config(materializedincremental,description订单明细表)}}SELECT*FROM{{ source(public,t_dwd_erp_salesorder)}} {%ifis_incremental()%}-- this filter will only be applied on an incremental runwherepodate(selectmax(podate)from{{ this }}){%endif%}当初始化或者重新生成这个模型dbt的运行参数为 dbt run –full-refresh时 is_incremental() 宏返回False从而省略 where 条件。但是在正常每天运行时is_incremental()则为True从而只选择增量数据。 ----参考自知乎。按照上述方式2.2.1.3.执行单个文件通过终端输出及数据库查询确定已经完成。2.2.3.视图 数据血缘题外话发现一个很逆天的东西这个玩意儿加分号会报错。虽然是sql文件但其实是因为做了文件关联它属于jinja文件所以不算传统sql。-- v_rpt_market.sql{{ config(materializedview,description市场分析报表)}}selectdim.cust_codeas客代,dim.cust_nameas客户名称,tag.area_typeas区域,coalesce(dim.province,dim.cust_cityname)as城市,sum(casewhent1.year2024thent1.totalamountrmbelse0end)as产品订单总金额_24年订单金额,sum(casewhent1.year2025thent1.totalamountrmbelse0end)as产品订单总金额_25年订单金额from(selectcustcode,sec_pro,extract(yearfrompodate)asyear,extract(monthfrompodate)asmonth,sum(totalamountrmb)totalamountrmbfromt_dwd_erp_salesordergroupbycustcode,sec_pro,extract(yearfrompodate),extract(monthfrompodate))t1leftjointagont1.custcodetag.cust_codeleftjoindimontag.cust_codedim.cust_codegroupbydim.cust_code,dim.cust_name,tag.area_type,dim.province,dim.cust_cityname直接找一个以前做的需求丢进去执行。# 单独执行该文件dbt run--modelsv_rpt_market.sql# 执行成功输出09:27:191of1START sql view model dbt_models.v_rpt_market............................[RUN]09:27:191of1OK created sql view model dbt_models.v_rpt_market.......................[CREATE VIEWin0.28s]09:27:19 Finished running1view modelin0hours0minutes and0.76seconds(0.76s). 09:27:19 09:27:19 Completed successfully成功生成视图注释名称是视图字段名。选择lineage发现只展示孤零零的v_rpt_market。丢给gemini将其优化为dbt可识别格式的SQL。{{ config(materializedview,description市场分析报表)}}-- 建议使用 CTE (WITH 语句) 结构这不仅能清晰血缘还方便调试withsales_ordersas(selectcustcode,sec_pro,extract(yearfrompodate)asyear,extract(monthfrompodate)asmonth,sum(totalamountrmb)astotalamountrmbfrom{{ ref(t_dwd_erp_salesorder)}}-- 关键点使用 ref 建立血缘wherepodate::date2024-01-01groupby1,2,3,4),tagsas(select*from{{ ref(tag)}}-- 使用 ref),dimas(select*from{{ ref(dim)}}-- 使用 ref)selectcd.cust_codeas客代,cd.cust_nameas客户名称,ct.area_typeas区域,coalesce(cd.province,cd.cust_cityname)as城市,-- 产品订单总金额sum(casewhent1.year2024thent1.totalamountrmbelse0end)as产品订单总金额_24年订单金额,sum(casewhent1.year2025thent1.totalamountrmbelse0end)as产品订单总金额_25年订单金额fromsales_orders t1leftjointags ctont1.custcodect.cust_codeleftjoindim cdonct.cust_codecd.cust_codegroupbycd.cust_code,cd.cust_name,ct.area_type,cd.province,cd.cust_cityname重新执行后再进入lineage就可以看见视图上游的表。2.2.4.Marco宏类似函数的东西先参考数据质量的自定义规则后续再对这个功能进行补充。2.3. 数据管理编写yml文件可以提供文档描述和数据质量检测还可以作为数据信息支持AI了解数据、结构进行下一步处理。2.3.1.生成数据文档DBT支持生成数据文档。未编写models.yml先分别执行以下代码dbt docs generate dbt docs serve弹出网页可以查看相关表的信息。但是没有注释不方便提供给下游使用。2.3.2.编写yml文件可以通过编写yml文件包括source.yml和models.yml为来源数据、转换数据添加注释、描述便于查阅管理。在数据库中已完成注释填充丢给gemini将其输出为dbt中models.yml可识别格式样例如下# models/models.ymlversion:2models:-name:t_dwd_erp_salesorderdescription:销售订单明细事实表columns:-name:reciddescription:记录 ID-name:deptnamedescription:三级部门保存后在后端重新执行以下代码# 需要先ctrlC结束原有文档dbt docs generate dbt docs serve效果如下除了展示外models.yml也是提交给AI应用部门的交付物。2.3.3.数据质量2.3.3.1.通用规则dbt提供四个通用规则可以直接调用。unique唯一not_null非空accepted_values在status模型中orders列的值应该为’placed’、‘shipped’、completed’或’returned’中的某一个。relationships: 在customer_id模型中每一个orders都相当于id中的customers。在yml文件中可以调用# models/models.ymlversion:2models: - name: t_dwd_erp_salesorder description:销售订单明细事实表columns: - name: autoid description:自增长 IDtests: - unique - not_null在数据文档中也会展示。如下图所示2.3.3.2.自定义规则官网声明自定义规则存储路径如下tests/generic/或macros/。在tests下新建文件夹generic并添加自定义规则# tests/generic/test_isnotzerp.sql{%test is_not_zero(model,column_name)%}withvalidationas(select{{ column_name }}ascheck_fieldfrom{{ model }}),validation_errorsas(selectcheck_fieldfromvalidation-- 核心逻辑找出所有等于 0 的记录wherecheck_field0)select*fromvalidation_errors {%endtest%}在yml文件中调用# models/models.ymlversion:2models: - name: t_dwd_erp_salesorder columns: - name: qtyorder# 假设订单数量不能为 0tests: - is_not_zer在终端执行测试及结果dbttest--selectt_dwd_erp_salesorder# 执行结果只展示自定义规则08:37:321of3STARTtestis_not_zero_t_dwd_erp_salesorder_qtyorder....................[RUN]08:37:321of3FAIL356is_not_zero_t_dwd_erp_salesorder_qtyorder......................[FAIL356in0.26s]可以看见确实可以执行自定义规则。但是按照全部历史数据检测而我并不需要我只需要对增量数据检测优化后更新为以下规则# tests/generic/test_isnotzerp.sql{%test is_not_zero_incremental(model,column_name,date_column,lookback_days3)%} {{ config(severitywarn)}}-- 直接在测试内部定义为告警withvalidationas(select{{ column_name }}ascheck_field,{{ date_column }}asevent_datefrom{{ model }}),validation_errorsas(selectcheck_fieldfromvalidationwherecheck_field0-- 核心逻辑只检查最近 N 天的数据andevent_datecurrent_date-interval{{ lookback_days }} day)select*fromvalidation_errors {%endtest%}yml文件也需要添加参数# models/models.ymlversion:2models:-name:t_dwd_erp_salesordercolumns:-name:qtyorder# 假设订单数量不能为 0tests:-is_not_zero_incremental:date_column:podate# 指定时间列lookback_days:90# 仅检查过去 90天内进入的数据刚好有一条不通过规则的数据再来执行在终端执行测试及结果dbttest--selectt_dwd_erp_salesorder# 结果如下09:28:071of3STARTtestis_not_zero_incremental_t_dwd_erp_salesorder_qtyorder__podate__90[RUN]09:28:071of3WARN1is_not_zero_incremental_t_dwd_erp_salesorder_qtyorder__podate__90[WARN1in0.26s]会有报黄效果测试只提醒、不报错等级改成报错{{ config(severityerror)}}-- 发现错误即报错中断流程效果如下可以在调度工作流中与数据更新搭配使用。2.3.3.3.执行执行测试代码如下# 运行所有数据测试dbttest--selecttest_type:data# 返回所有数据测试结果以下结果为截取示例07:22:061of6STARTtestnot_null_my_first_dbt_model_id...............................[RUN]07:22:061of6ERROR not_null_my_first_dbt_model_id....................................[ERRORin0.16s]07:22:062of6STARTtestnot_null_my_second_dbt_model_id..............................[RUN]07:22:062of6ERROR not_null_my_second_dbt_model_id...................................[ERRORin0.09s]07:22:063of6STARTtestnot_null_t_dwd_erp_salesorder_autoid.........................[RUN]07:22:073of6PASS not_null_t_dwd_erp_salesorder_autoid...............................[PASSin1.16s]07:22:074of6STARTtestunique_my_first_dbt_model_id.................................[RUN]07:22:074of6ERROR unique_my_first_dbt_model_id......................................[ERRORin0.07s]07:22:075of6STARTtestunique_my_second_dbt_model_id................................[RUN]07:22:085of6ERROR unique_my_second_dbt_model_id.....................................[ERRORin0.13s]07:22:086of6STARTtestunique_t_dwd_erp_salesorder_autoid...........................[RUN]07:22:106of6PASS unique_t_dwd_erp_salesorder_autoid.................................[PASSin2.49s]# 会有红色字体突出未通过数据质量检测的模型、数据质量检测类型# 例如07:22:11 Failureintestunique_my_first_dbt_model_id(models\example\schema.yml)07:22:11 Database Errorintestunique_my_first_dbt_model_id(models\example\schema.yml)relationdbt_models.my_first_dbt_modeldoes not exist LINE19: fromdb.dbt_models.my_first_dbt_model^ compiled code at target\run\chatkbi\models\example\schema.yml\unique_my_first_dbt_model_id.sql不想运行所有测试代码可以参考以下方法更复杂的方式参考官网测试示例# 1.仅该模型相关测试dbttest--selectmodelname# 运行该模型及其下游的所有测试dbttest--selectmodelname# 例如注意不用加后缀.sqldbttest--selectt_dwd_erp_salesorder# 2.仅源数据测试dbttest--selectsource:*# 3.仅进行一种校验例如仅唯一性校验dbttest--selecttest_type:unique还可以按标签执行测试。需要先有标签# models/models.ymlmodels:-name:orderscolumns:-name:order_idconfig:tags:[my_column_tag]# changed to config in v1.10 and backported to 1.9data_tests:-unique再运行以下代码# 仅运行带标签的测试dbttest--selecttag:my_column_tag标签也可以加载不同的地方甚至在test下例如models:-name:orderscolumns:-name:order_iddata_tests:-unique:config:tags:[my_test_tag]# changed to config in v1.102.3.3.4.对比gxgx是一款数据质量组件。由于国内文档较少比DBT少很多对其很多功能没有完全开发应用目前只使用SQL进行监控这一点其实DBT也能做到甚至更简洁方便。除此以外DBT的规则可以存储复用而GX的使用是制作模版套用替换DBT的规则在配置文件中体现更方便管理而GX目前是有记录期望值与实际值可以看见趋势对于维护来说提供更多信息。两者可以一起使用GX作为数据仓库的监控平台DBT的检测作为流向下游的第一道门槛。
【dbt】数据底座建设案例
发布时间：2026/7/16 10:29:13
【dbt】数据底座建设案例文章目录前言一、DBT二、使用步骤2.1. 项目初始化2.1.1.首次初始化2.1.2.中断初始化2.1.3.完成初始化2.1.4.检查是否完成配置2.2.创建模型2.2.1.全量 跨shcema取数说明2.2.1.1.方法1不使用source直接使用SQL语句完成。2.2.1.2.方法2使用source需要配置source.yml文件2.2.1.3.全量2.2.2.增量 : 首次全量非首次增量2.2.3.视图 数据血缘2.2.4.Marco宏2.3. 数据管理2.3.1.生成数据文档2.3.2.编写yml文件2.3.3.数据质量2.3.3.1.通用规则2.3.3.2.自定义规则2.3.3.3.执行2.3.3.4.对比gx前言在大模型与数据分析领域数据底座建设十分重要。接到需求需要对数仓已有明细进行整合写入新的目标库作为AI应用与数据分析的支撑数据。使用dbt完成包括以下内容交付数据库中的数据、使用dbt进行数据更新、dbt数据血缘、dbt形成文档、统计口径文档。一、DBT介绍详见官网。官网安装教程二、使用步骤本章使用的是vscode及插件power user for dbt。不使用插件dbt是因为dbt的数据血缘需要在网页中呈现power user for dbt可以在编译器窗口中展示。完成插件安装后需要在设置中添加一个文件关联# settings.jsonfiles.associations:{*.sql:jinja-sql},2.1. 项目初始化2.1.1.首次初始化# 在终端执行dbt init需要根据提示输入相关信息一般如下项目名称会帮你建立一个项目文件夹数据库类型默认是posrgrehostportusernamepassworddbname数据库名称填写目标数据库。schema模式名称填写目标模式名称。threads默认12.1.2.中断初始化建议准备好再执行和填写。试过填错中断ctrlC重新初始化无法用相同的项目名称。解决方法删掉该文件下的dbt_project.yml文件把整个文件夹删掉再次执行dbt init重新初始化。2.1.3.完成初始化完成后打开文件夹可以看到如下结构analysesdbt_packageslogsmacros存放可复用的 Jinja 宏类似 SQL 函数。models存放主要的SQL数据模型逻辑。seeds存放小的静态 CSV 文件如码表、映射表dbt 可以直接将其加载进数据库。snapshots存放用于记录数据历史变化的“快照”代码。targetdbt 编译后生成的 SQL 文件和运行结果存放的目录。test存放数据质量测试脚本。dbt_project.ymlREADME.md打开dbt_project.yml可以看到相关的配置信息。核心逻辑配置全局、局部设置物化存储的格式包括table表、view视图、incremental增量表也可以在具体的SQL文件中单独覆盖该全局设置。具体如下# yml文件models:materialized:table# 全局所有模型都是表最低filename:# 你的项目名称example:# 项目下的文件夹materialized:view# 局部特例只有 example 文件夹下的是视图中# SQL文件# 在SQL文件中设置最高{{ config(materialized...) }}{{config( materialized incremental )}}二编新增在dbt_project.yml中添加文档参数可以在dbt数据更新中把models.yml的注释同步到表中。models:filename:example:materialized:viewtables:materialized:tablepersist_docs:relation:true# 同步表注释columns:true# 同步列注释正常情况下还需要有profiles.yml、user.yml文件。这俩文件自动生成在根目录dbt文件下参考路径如下C:\Users\用户名称\.dbt打开profiles.yml可以看见初始化时填写的信息记录在该文件中。关于该配置文件如有建立建议复制一份在项目根目录下方便后续文件打包与查阅.user.yml也建议复制一份。并且重新初始化会覆盖原有配置信息profiles.yml文件属性如下图如果没有生成PG可以参考如下格式新建profiles.yml文件参考自知乎其他数据库可查找官网profiles.yml的说明文档。default:target:devoutputs:dev:type:postgreshost:[hostname]user:[username]password:[password]port:[port]dbname:[database name]schema:[dbt schema]threads:[1 or more]keepalives_idle:0# default 0, indicating the system defaultsearch_path:[optional,override the default postgres search_path]role:[optional,set the role dbt assumes when executing queries]sslmode:[optional,set the sslmode used to connect to the database]2.1.4.检查是否完成配置在终端项目文件下执行以下命令# 执行命令dbt debug# 执行结果09:38:23 All checks passed!# 配置完成可以看到使用的配置文件路径使用的是根目录路径下的还是项目目录下的profiles.yml2.2.创建模型完成配置可以开始创建模型。我在models文件夹下分别创建文件夹tables、views不把创建模型放在example中。在models/tables文件夹下创建sql文件。2.2.1.全量 跨shcema取数说明注意1profiles.yml中只配置了目标数据库而没有要求配置来源数据库。是由于DBT默认数据已完成集成仅支持从Schema A迁移转化至Schema B。因此需要确保配置文件中的数据库为来源数据库并且也是目标数据库配置文件中的schema为目标shcema后续步骤为跨schema取数、转换、写入。注意2sql文件名就是表名。2.2.1.1.方法1不使用source直接使用SQL语句完成。-- t_dim_date.sql{{ config(materializedtable,description日期维度表)}}SELECT*FROMpublic.t_dim_date2.2.1.2.方法2使用source需要配置source.yml文件sourece.yml需要创建在models路径下。# models/sources.ymlversion:2sources:-name:publicschema:public# 确保这个 schema 在你连接的那个库里存在tables:-name:t_dim_date-- t_dim_date.sql{{ config(materializedtable,description日期维度表)}}SELECT*FROM{{ source(public,t_dim_date)}}2.2.1.3.全量不需要在代码里写TRUNCATE或DELETE。在 dbt 中只要物化方式是table每一次dbt run都是一次全量覆盖更新。你只需要在 SQL 中控制生成的日期范围例如每年年初把 SQL 里的结束日期往后改一年然后重新运行即可。# 运行整个dbt项目dbt run# 运行单个文件dbt run--modlesfilename# 例如dbt run--modelst_dim_date.sql# 成功返回内容07:22:251of1START sql table model dbt_models.t_dim_date.............................[RUN]07:22:251of1OK created sql table model dbt_models.t_dim_date........................[SELECT1096in0.41s]07:22:25 07:22:25 Finished running1table modelin0hours0minutes and1.15seconds(1.15s). 07:22:25 07:22:25 Completed successfully这时候去对应的数据库schema中查看可以查到该表。2.2.2.增量 : 首次全量非首次增量{{ config(materializedincremental,description订单明细表)}}SELECT*FROM{{ source(public,t_dwd_erp_salesorder)}} {%ifis_incremental()%}-- this filter will only be applied on an incremental runwherepodate(selectmax(podate)from{{ this }}){%endif%}当初始化或者重新生成这个模型dbt的运行参数为 dbt run –full-refresh时 is_incremental() 宏返回False从而省略 where 条件。但是在正常每天运行时is_incremental()则为True从而只选择增量数据。 ----参考自知乎。按照上述方式2.2.1.3.执行单个文件通过终端输出及数据库查询确定已经完成。2.2.3.视图 数据血缘题外话发现一个很逆天的东西这个玩意儿加分号会报错。虽然是sql文件但其实是因为做了文件关联它属于jinja文件所以不算传统sql。-- v_rpt_market.sql{{ config(materializedview,description市场分析报表)}}selectdim.cust_codeas客代,dim.cust_nameas客户名称,tag.area_typeas区域,coalesce(dim.province,dim.cust_cityname)as城市,sum(casewhent1.year2024thent1.totalamountrmbelse0end)as产品订单总金额_24年订单金额,sum(casewhent1.year2025thent1.totalamountrmbelse0end)as产品订单总金额_25年订单金额from(selectcustcode,sec_pro,extract(yearfrompodate)asyear,extract(monthfrompodate)asmonth,sum(totalamountrmb)totalamountrmbfromt_dwd_erp_salesordergroupbycustcode,sec_pro,extract(yearfrompodate),extract(monthfrompodate))t1leftjointagont1.custcodetag.cust_codeleftjoindimontag.cust_codedim.cust_codegroupbydim.cust_code,dim.cust_name,tag.area_type,dim.province,dim.cust_cityname直接找一个以前做的需求丢进去执行。# 单独执行该文件dbt run--modelsv_rpt_market.sql# 执行成功输出09:27:191of1START sql view model dbt_models.v_rpt_market............................[RUN]09:27:191of1OK created sql view model dbt_models.v_rpt_market.......................[CREATE VIEWin0.28s]09:27:19 Finished running1view modelin0hours0minutes and0.76seconds(0.76s). 09:27:19 09:27:19 Completed successfully成功生成视图注释名称是视图字段名。选择lineage发现只展示孤零零的v_rpt_market。丢给gemini将其优化为dbt可识别格式的SQL。{{ config(materializedview,description市场分析报表)}}-- 建议使用 CTE (WITH 语句) 结构这不仅能清晰血缘还方便调试withsales_ordersas(selectcustcode,sec_pro,extract(yearfrompodate)asyear,extract(monthfrompodate)asmonth,sum(totalamountrmb)astotalamountrmbfrom{{ ref(t_dwd_erp_salesorder)}}-- 关键点使用 ref 建立血缘wherepodate::date2024-01-01groupby1,2,3,4),tagsas(select*from{{ ref(tag)}}-- 使用 ref),dimas(select*from{{ ref(dim)}}-- 使用 ref)selectcd.cust_codeas客代,cd.cust_nameas客户名称,ct.area_typeas区域,coalesce(cd.province,cd.cust_cityname)as城市,-- 产品订单总金额sum(casewhent1.year2024thent1.totalamountrmbelse0end)as产品订单总金额_24年订单金额,sum(casewhent1.year2025thent1.totalamountrmbelse0end)as产品订单总金额_25年订单金额fromsales_orders t1leftjointags ctont1.custcodect.cust_codeleftjoindim cdonct.cust_codecd.cust_codegroupbycd.cust_code,cd.cust_name,ct.area_type,cd.province,cd.cust_cityname重新执行后再进入lineage就可以看见视图上游的表。2.2.4.Marco宏类似函数的东西先参考数据质量的自定义规则后续再对这个功能进行补充。2.3. 数据管理编写yml文件可以提供文档描述和数据质量检测还可以作为数据信息支持AI了解数据、结构进行下一步处理。2.3.1.生成数据文档DBT支持生成数据文档。未编写models.yml先分别执行以下代码dbt docs generate dbt docs serve弹出网页可以查看相关表的信息。但是没有注释不方便提供给下游使用。2.3.2.编写yml文件可以通过编写yml文件包括source.yml和models.yml为来源数据、转换数据添加注释、描述便于查阅管理。在数据库中已完成注释填充丢给gemini将其输出为dbt中models.yml可识别格式样例如下# models/models.ymlversion:2models:-name:t_dwd_erp_salesorderdescription:销售订单明细事实表columns:-name:reciddescription:记录 ID-name:deptnamedescription:三级部门保存后在后端重新执行以下代码# 需要先ctrlC结束原有文档dbt docs generate dbt docs serve效果如下除了展示外models.yml也是提交给AI应用部门的交付物。2.3.3.数据质量2.3.3.1.通用规则dbt提供四个通用规则可以直接调用。unique唯一not_null非空accepted_values在status模型中orders列的值应该为’placed’、‘shipped’、completed’或’returned’中的某一个。relationships: 在customer_id模型中每一个orders都相当于id中的customers。在yml文件中可以调用# models/models.ymlversion:2models: - name: t_dwd_erp_salesorder description:销售订单明细事实表columns: - name: autoid description:自增长 IDtests: - unique - not_null在数据文档中也会展示。如下图所示2.3.3.2.自定义规则官网声明自定义规则存储路径如下tests/generic/或macros/。在tests下新建文件夹generic并添加自定义规则# tests/generic/test_isnotzerp.sql{%test is_not_zero(model,column_name)%}withvalidationas(select{{ column_name }}ascheck_fieldfrom{{ model }}),validation_errorsas(selectcheck_fieldfromvalidation-- 核心逻辑找出所有等于 0 的记录wherecheck_field0)select*fromvalidation_errors {%endtest%}在yml文件中调用# models/models.ymlversion:2models: - name: t_dwd_erp_salesorder columns: - name: qtyorder# 假设订单数量不能为 0tests: - is_not_zer在终端执行测试及结果dbttest--selectt_dwd_erp_salesorder# 执行结果只展示自定义规则08:37:321of3STARTtestis_not_zero_t_dwd_erp_salesorder_qtyorder....................[RUN]08:37:321of3FAIL356is_not_zero_t_dwd_erp_salesorder_qtyorder......................[FAIL356in0.26s]可以看见确实可以执行自定义规则。但是按照全部历史数据检测而我并不需要我只需要对增量数据检测优化后更新为以下规则# tests/generic/test_isnotzerp.sql{%test is_not_zero_incremental(model,column_name,date_column,lookback_days3)%} {{ config(severitywarn)}}-- 直接在测试内部定义为告警withvalidationas(select{{ column_name }}ascheck_field,{{ date_column }}asevent_datefrom{{ model }}),validation_errorsas(selectcheck_fieldfromvalidationwherecheck_field0-- 核心逻辑只检查最近 N 天的数据andevent_datecurrent_date-interval{{ lookback_days }} day)select*fromvalidation_errors {%endtest%}yml文件也需要添加参数# models/models.ymlversion:2models:-name:t_dwd_erp_salesordercolumns:-name:qtyorder# 假设订单数量不能为 0tests:-is_not_zero_incremental:date_column:podate# 指定时间列lookback_days:90# 仅检查过去 90天内进入的数据刚好有一条不通过规则的数据再来执行在终端执行测试及结果dbttest--selectt_dwd_erp_salesorder# 结果如下09:28:071of3STARTtestis_not_zero_incremental_t_dwd_erp_salesorder_qtyorder__podate__90[RUN]09:28:071of3WARN1is_not_zero_incremental_t_dwd_erp_salesorder_qtyorder__podate__90[WARN1in0.26s]会有报黄效果测试只提醒、不报错等级改成报错{{ config(severityerror)}}-- 发现错误即报错中断流程效果如下可以在调度工作流中与数据更新搭配使用。2.3.3.3.执行执行测试代码如下# 运行所有数据测试dbttest--selecttest_type:data# 返回所有数据测试结果以下结果为截取示例07:22:061of6STARTtestnot_null_my_first_dbt_model_id...............................[RUN]07:22:061of6ERROR not_null_my_first_dbt_model_id....................................[ERRORin0.16s]07:22:062of6STARTtestnot_null_my_second_dbt_model_id..............................[RUN]07:22:062of6ERROR not_null_my_second_dbt_model_id...................................[ERRORin0.09s]07:22:063of6STARTtestnot_null_t_dwd_erp_salesorder_autoid.........................[RUN]07:22:073of6PASS not_null_t_dwd_erp_salesorder_autoid...............................[PASSin1.16s]07:22:074of6STARTtestunique_my_first_dbt_model_id.................................[RUN]07:22:074of6ERROR unique_my_first_dbt_model_id......................................[ERRORin0.07s]07:22:075of6STARTtestunique_my_second_dbt_model_id................................[RUN]07:22:085of6ERROR unique_my_second_dbt_model_id.....................................[ERRORin0.13s]07:22:086of6STARTtestunique_t_dwd_erp_salesorder_autoid...........................[RUN]07:22:106of6PASS unique_t_dwd_erp_salesorder_autoid.................................[PASSin2.49s]# 会有红色字体突出未通过数据质量检测的模型、数据质量检测类型# 例如07:22:11 Failureintestunique_my_first_dbt_model_id(models\example\schema.yml)07:22:11 Database Errorintestunique_my_first_dbt_model_id(models\example\schema.yml)relationdbt_models.my_first_dbt_modeldoes not exist LINE19: fromdb.dbt_models.my_first_dbt_model^ compiled code at target\run\chatkbi\models\example\schema.yml\unique_my_first_dbt_model_id.sql不想运行所有测试代码可以参考以下方法更复杂的方式参考官网测试示例# 1.仅该模型相关测试dbttest--selectmodelname# 运行该模型及其下游的所有测试dbttest--selectmodelname# 例如注意不用加后缀.sqldbttest--selectt_dwd_erp_salesorder# 2.仅源数据测试dbttest--selectsource:*# 3.仅进行一种校验例如仅唯一性校验dbttest--selecttest_type:unique还可以按标签执行测试。需要先有标签# models/models.ymlmodels:-name:orderscolumns:-name:order_idconfig:tags:[my_column_tag]# changed to config in v1.10 and backported to 1.9data_tests:-unique再运行以下代码# 仅运行带标签的测试dbttest--selecttag:my_column_tag标签也可以加载不同的地方甚至在test下例如models:-name:orderscolumns:-name:order_iddata_tests:-unique:config:tags:[my_test_tag]# changed to config in v1.102.3.3.4.对比gxgx是一款数据质量组件。由于国内文档较少比DBT少很多对其很多功能没有完全开发应用目前只使用SQL进行监控这一点其实DBT也能做到甚至更简洁方便。除此以外DBT的规则可以存储复用而GX的使用是制作模版套用替换DBT的规则在配置文件中体现更方便管理而GX目前是有记录期望值与实际值可以看见趋势对于维护来说提供更多信息。两者可以一起使用GX作为数据仓库的监控平台DBT的检测作为流向下游的第一道门槛。
