创建数据基础模块

虽然 Cortex Framework 为 SAP (cortex.sap) 等企业 ERP 系统提供了开箱即用的 数据基础 模块，但您也可以在 自定义命名空间 内创建新的自定义数据基础模块。这样，您就可以定义自定义构建行为，并将支持范围扩展到新的源系统。这些系统可以包括 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 数据集内的一致性记录。

如需验证自定义数据基础模块是否成功编译和部署，请参阅数据产品可扩展性页面中的验证部分。