dbt 自動化檢查 - Generator - iT 邦幫忙::一起幫忙解決難題，拯救 IT 人的一天

2024 iThome 鐵人賽

DAY 22

AI/ ML & Data

dbt 修煉之路系列第 22 篇

dbt 自動化檢查 - Generator

16th鐵人賽

joshua_lin

團隊資料工程師甘苦談

2024-10-06 15:05:38

81 瀏覽

分享至

在介紹架構的文章中有提到 Generator 是負責動態產生 PR 報告檔案，本篇的內文將會詳細介紹如何使用Jinja模板和Python來動態生成 PR 檢查報告。過程包括定義模板、準備數據和生成最終的HTML報告。

Jinja 模板設計

在開始自動化檢查流程時，我們首先確定了需要檢查的項目，包括：

欄位增減
型態改變
資料筆數變化
相關exposures
停用models
異動macros相關models
孤兒物件

基於這些需求，我們使用Jinja設計了一個靈活的HTML模板。這個模板能夠根據每個PR的具體檢查結果，動態生成適當的表格或列表。

模板結構

模板主要分為以下幾個部分：

報告標題
相關 exposures 列表
停用 models 列表
異動 macros 相關 models 列表
孤兒物件表格
欄位增減檢查表格
欄位型態檢查表格（異動models）
欄位型態檢查表格（下游表）：下游表指的是有使用 ”更改 models“ 作為上游表的 models，上游表異動可能會造成下游表欄位改變
資料筆數檢查表格

以下是 Jinja 模板

<!DOCTYPE html>
<h2>[PR 檢查報告]</h2>
{% if exposure_list %}
<h3>相關 exposures: </h3>
{% for exposure in exposure_list %}
<ul>
  <li><a href='{{ exposure.url }}'>{{ exposure.label }}</a></li>
</ul>
{% endfor %}
{% endif %}

{% if disabled_list %}
<h3>停用 models(未在 manifest 中找到): </h3>
{% for diabled in disabled_list %}
<ul>
  <li>{{ diabled }}</li>
</ul>
{% endfor %}
{% endif %}

{% if macro_related_list %}
<h3>異動 macros 相關 models: </h3>
{% for macro_related in macro_related_list %}
<ul>
  {% if macro_related.materialized == "incremental" %}
  <li>{{ macro_related.model }}({{ macro_related.materialized }})</li>
  {% else %}
  <li>{{ macro_related.model }}</li>
  {% endif %}
</ul>
{% endfor %}
{% endif %}

{% if orphan_result %}
<h3>孤兒物件(存在 BQ 但不存在於 dbt models，Merge 時會自動刪除！): </h3>
<table style='width:100%'>
  <tr style='text-align: left;'>
    <th>資料來源</th>
    <th>類別</th>
    <th>表名</th>
  </tr>
  {% for row in orphan_result %}
  <tr>
    <td>{{ row.schema_name }}</td>
    <td>{{ row.table_type }}</td>
    <td>{{ row.table_name }}</td>
  </tr>
  {% endfor %}
</table><br>
{% endif %}

{% if column_diff_result %}
<h3>欄位增減檢查: </h3>
<table style='width:100%'>
  <tr style='text-align: left;'>
    <th>資料來源</th>
    <th>表名</th>
    <th>欄位增減</th>
    <th>欄位名</th>
  </tr>
  {% for row in column_diff_result %}
  <tr>
    <td>{{ row.schema_name }}</td>
    <td>{{ row.table_name }}</td>
    <td>{{ row.changed_type }}</td>
    <td>{{ row.column_name }}</td>
  </tr>
  {% endfor %}
</table><br>

{% else %}
<h3>欄位增減檢查: Pass</h3>
{% endif %}

{% if column_type_diff_result %}
<h3>欄位型態檢查(異動 models): </h3>
<table style='width:100%'>
  <tr style='text-align: left;'>
    <th>資料來源</th>
    <th>表名</th>
    <th>欄位名</th>
    <th>欄位型態變化</th>
  </tr>
  {% for row in column_type_diff_result %}
  <tr>
    <td>{{ row.schema_name }}</td>
    <td>{{ row.table_name }}</td>
    <td>{{ row.column_name }}</td>
    <td>{{ row.type_changed }}</td>
  </tr>
  {% endfor %}
</table><br>

{% else %}
<h3>欄位型態檢查(異動 models): Pass</h3>
{% endif %}

{% if downstream_column_type_diff_result %}
<h3>欄位型態檢查(下游表): </h3>
<table style='width:100%'>
  <tr style='text-align: left;'>
    <th>資料來源</th>
    <th>表名</th>
    <th>欄位名</th>
    <th>欄位型態變化</th>
  </tr>
  {% for row in downstream_column_type_diff_result %}
  <tr>
    <td>{{ row.schema_name }}</td>
    <td>{{ row.table_name }}</td>
    <td>{{ row.column_name }}</td>
    <td>{{ row.type_changed }}</td>
  </tr>
  {% endfor %}
</table><br>

{% else %}
<h3>欄位型態檢查(下游表): Pass</h3>
{% endif %}

{% if row_count_diff_result %}
<h3>資料筆數檢查: </h3>
<table style='width:100%'>
  <tr style='text-align: left;'>
    <th>資料來源</th>
    <th>表名</th>
    <th>異動前筆數</th>
    <th>異動後筆數</th>
    <th>異動百分比</th>
  </tr>
  {% for row in row_count_diff_result %}
  <tr>
    <td>{{ row.schema_name }}</td>
    <td>{{ row.table_name }}</td>
    <td>{{ row.master_row_count }}</td>
    <td>{{ row.branch_row_count }}</td>
    <td>{{ row.diff_pct }}</td>
  </tr>
  {% endfor %}
</table><br>

{% else %}
<h3>資料筆數檢查: Pass</h3>
{% endif %}

Python 生成 HTML 報告

定義好 Jinja 模板後，我們需要使用 Python 將檢查結果傳入模板，從而生成最終的 HTML 報告。以下是實現這一功能的 HTMLGenerator：

import os
import logging
from typing import Dict
from jinja2 import Environment, FileSystemLoader, Template

class HTMLGenerator:
    def __init__(self, templates_dir):
        current_dir = os.path.dirname(os.path.abspath(__file__))
        self.templates_dir = os.path.join(current_dir, templates_dir)
        if os.path.exists(self.templates_dir):
            self.env = Environment(loader=FileSystemLoader(self.templates_dir))
        else:
            msg = f"Can not find template directory in: {self.templates_dir}"
            logging.error(msg)
            raise FileNotFoundError(msg)

    def generate_html(self, template_name: str, data: Dict) -> str:
        template = self._get_template(template_name)
        # 套用模板
        return template.render(data)

    def _get_template(self, template_name: str) -> Template:
        try:
            return self.env.get_template(template_name)
        except Exception as e:
            logging.error(f"Can not find {template_name} in {self.templates_dir}")
            raise e

以上就是 generators 的完整介紹和程式碼，下一篇將會介紹 pr-check 產生發送報告資料的模組 - operators