建立資料基礎模組

Cortex Framework 提供適用於企業 ERP 系統 (例如 SAP (cortex.sap)) 的現成資料基礎模組，您也可以在自訂命名空間中建立新的自訂資料基礎模組。您可以藉此定義自訂建構行為，並將支援範圍擴展至新的來源系統。例如 PostgreSQL、MySQL 等資料庫管理系統，會將原始資料複製到 BigQuery。

本指南將逐步說明如何為客戶服務單系統建立新的資料基礎模組，該系統的資料 (customers、tickets、ticketlogitem 資料表) 已從 PostgreSQL 資料庫複製到名為 ticketing_data_raw 的原始 BigQuery 資料集。

建立自訂資料基礎模組時，建議使用專屬自訂命名空間，將擴充功能和自訂項目與 Cortex Framework 構件分開，藉此提升生命週期管理效率。

情境範例總覽

在本逐步操作說明中，我們將：

  1. 建立專屬的自訂命名空間 ticketing，隔離資料基礎資產。
  2. 定義路徑 ticketing.ticketing.foundations.ticketing_system 的新資料基礎模組。
  3. 設定 customers、tickets 和 ticketlogitem 資料表的資料表設定 (table_settings.default.yaml)。
  4. 為每個資料表建立欄和欄位層級的中繼資料註解。
  5. 在 config/config.yaml 中註冊原始資料來源 (ticketing_data_raw)、目標一致性資料集 (data_foundation_ticketing) 和新的基礎模組。

模組資料夾和檔案結構

新資料基礎模組的所有實體檔案都位於 src/data_modules/ 下的自訂命名空間中。下表和目錄樹狀結構概述了每個檔案的放置位置：

config/
└── config.yaml                                           # Global configuration & module registration
src/data_modules/ticketing/ticketing/foundations/ticketing_system/
├── manifest.yaml                                         # Declares module category, type, and builder
├── table_settings.default.yaml                           # Table materialization, bigQueryLabels, dataformTags, and layouts
├── builder.py                                            # (Optional) Custom Dataform generator class for this module
└── annotations/                                          # Field and column-level schema descriptions
    ├── customers.yaml
    ├── tickets.yaml
    └── ticketlogitem.yaml

重要事項：開始前，請確認您打算處理的來源資料表位於原始層資料集。

檔案或目錄路徑 用途和說明
config/config.yaml 註冊 ticketing 命名空間、PostgreSQL 原始資料來源、目標 BigQuery 資料集和基礎模組例項。
src/data_modules/ticketing/ticketing/foundations/ticketing_system/manifest.yaml 宣告模組中繼資料、顯示名稱、類別 (例如 foundation)、模組類型 (例如 generic)，以及編譯期間使用的產生器建構工具類別。
src/data_modules/ticketing/ticketing/foundations/ticketing_system/table_settings.default.yaml 設定要從 ticketing_data_raw 調整哪些來源資料表，以及 BigQuery 最佳化 dataformTags、bigQueryLabels、分割區詳細資料和叢集詳細資料。
src/data_modules/ticketing/ticketing/foundations/ticketing_system/annotations/*.yaml 內含豐富的 YAML 中繼資料，說明資料表和欄位定義。這些內容會自動合併至已編譯的 Dataform 定義，因此說明會保留在 BigQuery 資料表的中繼資料中。
src/data_modules/ticketing/ticketing/foundations/ticketing_system/builder.py 選填。如果來源資料庫在編譯期間需要自訂資料清理作業或特定方言的 SQL 轉換作業，您可以在這裡定義模組層級的建構工具類別。

步驟 1：在 config.yaml 中註冊命名空間、資料來源和目標

建立實體檔案前，請開啟部署設定檔 (config/config.yaml)，並宣告自訂命名空間、原始 PostgreSQL 來源資料集，以及要建立相符資料表的目的地資料集：

data:
  namespaces:
    - name: cortex
      path: ../src/data_modules/cortex
    - name: ticketing                              # <-- Name of custom namespace
      path: ../src/data_modules/ticketing          # <-- Points to subdirectory under 'src/data_modules/'

  datasets:
    - id: ticketing_data_raw                       # <-- Unique source ID
      projectId: "source_project_id"
      datasetId: ticketing_data_raw                # <-- Raw dataset containing PostgreSQL replication tables
    - id: data_foundation_ticketing                # <-- Unique target ID
      projectId: "target_project_id"
      datasetId: data_foundation_ticketing         # <-- Target dataset for conformed foundation tables

步驟 2：在 config.yaml 中註冊資料基礎模組

在 config/config.yaml 的 data.modules.foundations 區段下方，註冊新的資料基礎模組例項，將資料來源 (ticketing_data_raw) 連結至資料目標 (data_foundation_ticketing)：

data:
  modules:
    foundations:
      - moduleId: ticketing_foundation
        modulePath: ticketing.ticketing.foundations.ticketing_system   # Format: {namespace}.{systemtype}.{module_type:foundations}.{subsystemtype}
        dataSourceId: ticketing_data_raw
        dataTargetId: data_foundation_ticketing
        # Custom table settings file relative to 'config/' directory
        # Recommended path: '{namespace_dir}/{system_type}/foundations/{system_sub_type}/table_settings.yaml'
        # If omitted, defaults to "../src/data_modules/ticketing/ticketing/foundations/ticketing_system/table_settings.default.yaml"
        tableSettings: "ticketing/ticketing/foundations/ticketing_system/table_settings.yaml"

步驟 3：建立模組資訊清單檔案

建立資訊清單檔案，宣告模組中繼資料：src/data_modules/ticketing/ticketing/foundations/ticketing_system/manifest.yaml。

displayName: Ticketing System Data Foundation
description: Conformed foundation tables for PostgreSQL raw ticketing database.
category: foundation
type: generic
builder: ticketing_foundation

步驟 4：建立資料表設定檔 (table_settings.default.yaml)

建立預設資料表設定檔 src/data_modules/ticketing/ticketing/foundations/ticketing_system/table_settings.default.yaml。這個檔案會定義如何在 BigQuery 中具體化、分割及叢集化複製的 PostgreSQL 資料表 (customers、tickets、ticketlogitem)：

common:
  - source:
      tableName: customers
    target:
      bigQueryLabels:
        - key: data_class
          value: master
      dataformTags: [ticketing, foundation, masterdata]
      clusterDetails:
        columns: [customer_id]

  - source:
      tableName: tickets
    target:
      bigQueryLabels:
        - key: data_class
          value: transactional
      dataformTags: [ticketing, foundation, transactional]
      partitionDetails:
        column: created_at
        partitionType: time
        timeGrain: day
      clusterDetails:
        columns: [ticket_id, customer_id]

  - source:
      tableName: ticketlogitem
    target:
      bigQueryLabels:
        - key: data_class
          value: transactional
      dataformTags: [ticketing, foundation, transactional]
      partitionDetails:
        column: log_timestamp
        partitionType: time
        timeGrain: day
      clusterDetails:
        columns: [ticket_id, log_id]

步驟 5：建立欄位層級的中繼資料註解

為確保 BigQuery 中符合規範的資料表含有清楚的說明文件，請在 src/data_modules/ticketing/ticketing/foundations/ticketing_system/annotations/ 內為每個資料表建立註解 YAML 檔案。檔案名稱必須與來源表格名稱完全一致。

annotations/customers.yaml

description: "Customer master data conformed from PostgreSQL raw ticketing database."
fields:
  - name: "customer_id"
    description: "Unique customer identifier, PK"
  - name: "email"
    description: "Primary email address associated with the customer"
  - name: "full_name"
    description: "Customer full name or account contact name"
  - name: "created_at"
    description: "Timestamp when the customer record was originally created in PostgreSQL"

annotations/tickets.yaml

description: "Customer service tickets conformed from PostgreSQL raw ticketing database."
fields:
  - name: "ticket_id"
    description: "Unique ticket identifier, PK"
  - name: "customer_id"
    description: "Foreign key referencing customers.customer_id"
  - name: "subject"
    description: "Summary or subject line of the customer inquiry"
  - name: "status"
    description: "Current ticket lifecycle status (e.g., OPEN, IN_PROGRESS, RESOLVED, CLOSED)"
  - name: "priority"
    description: "Priority severity level (e.g., LOW, MEDIUM, HIGH, URGENT)"
  - name: "created_at"
    description: "Timestamp when the ticket was created"
  - name: "updated_at"
    description: "Timestamp when the ticket was last modified"

annotations/ticketlogitem.yaml

description: "Audit log history and activity events for customer service tickets."
fields:
  - name: "log_id"
    description: "Unique log event identifier, PK"
  - name: "ticket_id"
    description: "Foreign key referencing tickets.ticket_id"
  - name: "action"
    description: "Action or event performed on the ticket"
  - name: "description"
    description: "Notes and description on performed events on the ticket"
  - name: "performed_by"
    description: "User, agent, or automated system that performed the action"
  - name: "log_timestamp"
    description: "Exact timestamp when the activity log event occurred"

步驟 6：(選用) 定義自訂基礎建構工具

如果 PostgreSQL 資料基礎在編譯期間需要邏輯 (例如自動轉換資料類型、時間戳記轉換，或所有資料表的資料清理規則)，您可以定義這個模組範圍的自訂建構工具。

建立 src/data_modules/ticketing/ticketing/foundations/ticketing_system/builder.py：

import logging
import pathlib
import yaml
from common.builders.base import FoundationBuilder, Source
from common.registry import builder_registry
from common.schemas import config_schema, manifest_schema

logger = logging.getLogger(__name__)

@builder_registry.register("ticketing_foundation")
class TicketingFoundationBuilder(FoundationBuilder[config_schema.BaseModuleConfig]):
    """Custom Dataform generator for PostgreSQL ticketing data foundation."""

    def build(
        self,
        *,
        module_id: str,
        module_config: config_schema.BaseModuleConfig,
        global_config: config_schema.GlobalConfig,
        manifest: manifest_schema.ManifestConfig,
        base_dir: pathlib.Path,
        annotations_dir: pathlib.Path,
        output_dir: pathlib.Path,
        module_dir_name: str,
        sources_registry: set[Source],
        table_settings_file: pathlib.Path | None = None,
        required_tables: set[str] | None = None,
    ) -> None:
        logger.info("Building ticketing data foundation for module: %s", module_id)
        
        # 1. Load table settings
        if not table_settings_file or not table_settings_file.exists():
            logger.warning("No valid table settings found for %s", module_id)
            return

        with open(table_settings_file, encoding="utf-8") as f:
            settings = yaml.safe_load(f) or {}

        tables = settings.get("common", [])
        source_config = global_config.get_data_source(module_config.data_source_id)
        target_dataset = global_config.get_data_target(module_config.data_target_id)

        # 2. Generate Dataform .sqlx files for each table
        for table_item in tables:
            source_table = table_item["source"]["tableName"]
            if required_tables and source_table not in required_tables and not table_item.get("deployAlways"):
                continue

            # Register source table for centralized source generation
            sources_registry.add(Source(source_config.project_id, source_config.dataset_id, source_table))

            # Retrieve labels if configured
            bigquery_config = {}
            if "bigQueryLabels" in table_item["target"]:
                labels_dict = {label["key"]: label["value"] for label in table_item["target"]["bigQueryLabels"]}
                bigquery_config["labels"] = labels_dict

            dataform_tags = table_item["target"].get("dataformTags", ["ticketing", "foundation"])

            sqlx_content = f"""config {{
  type: "table",
  schema: "{target_dataset.dataset_id}",
  name: "{source_table}",
  tags: {dataform_tags}"""
            
            if bigquery_config:
                sqlx_content += f",\n  bigquery: {bigquery_config}"
                
            sqlx_content += f"""
}}

SELECT *
FROM `${{source_config.project_id}}.${{source_config.dataset_id}}.{source_table}`
"""
            out_file = output_dir / f"{source_table}.sqlx"
            out_file.write_text(sqlx_content, encoding="utf-8")
            logger.info("Generated %s", out_file)

驗證新的基礎模組

如要驗證及部署新建立的資料基礎模組，請按照下列步驟操作：

  1. 執行 Cortex Framework 建構和部署指令碼： bash uv run cortex-build-and-deploy --config "config/config.yaml"
  2. 確認 Dataform 編譯作業成功完成，且沒有發生錯誤，並確認系統已為 customers、tickets 和 ticketlogitem 生成 .sqlx 指令碼。
  3. 按照部署後步驟執行 Dataform 管道動作，並在 BigQuery 的 data_foundation_ticketing 資料集中驗證符合規範的記錄。

如要確認自訂資料基礎模組是否編譯及部署成功，請參閱資料產品擴充性頁面的「驗證」部分。