创建数据基础模块
虽然 Cortex Framework 为 SAP (cortex.sap) 等企业 ERP 系统提供了开箱即用的 数据基础 模块,但您也可以在 自定义命名空间 内创建新的自定义数据基础模块。这样,您就可以定义自定义构建行为,并将支持范围扩展到新的源系统。这些系统可以包括 PostgreSQL、MySQL 等数据库管理系统,它们会将原始数据复制到 BigQuery 中。
本指南将通过一个端到端示例,介绍如何为客户服务工单系统 创建新的数据基础模块,该系统的数据(customers、tickets、ticketlogitem 表)已从 PostgreSQL 数据库复制到名为 ticketing_data_raw 的原始 BigQuery 数据集中。
创建自定义 数据基础模块时,我们建议使用专用的 自定义命名空间,以便将扩展程序和自定义项与 Cortex Framework 工件分开,从而改进生命周期管理。
示例场景概览
在本演练中,我们将:
- 创建专用自定义命名空间
ticketing,以隔离数据基础资产。 - 定义路径为
ticketing.ticketing.foundations.ticketing_system的新数据基础模块。 - 为
customers、tickets和ticketlogitem表配置表设置 (table_settings.default.yaml)。 - 为每个表创建列级和字段级元数据注解。
- 在
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)
验证新的基础模块
如需验证和部署新创建的数据基础模块,请执行以下操作:
- 执行 Cortex Framework 构建和部署脚本:
bash uv run cortex-build-and-deploy --config "config/config.yaml" - 检查 Dataform 编译是否成功且没有错误,以及是否为
customers、tickets和ticketlogitem生成了.sqlx脚本。 - 按照部署后步骤运行 Dataform 流水线操作,并验证 BigQuery 中
data_foundation_ticketing数据集内的一致性记录。
如需验证自定义数据基础模块是否成功编译和部署,请参阅数据产品可扩展性页面中的验证部分。