Erstellung eines Datenfundierungsmoduls

Cortex Framework bietet sofort einsatzbereite Datenfundierungs-Module für ERP-Systeme von Unternehmen wie SAP (cortex.sap). Sie können aber auch neue benutzerdefinierte Datenfundierungsmodule in einem benutzerdefinierten Namespace erstellen. So können Sie benutzerdefinierte Build-Verhaltensweisen definieren und die Unterstützung auf ein neues Quellsystem ausweiten. Dazu können Datenbankverwaltungssysteme wie PostgreSQL, MySQL usw. gehören, die ihre Rohdaten in BigQuery replizieren.

In dieser Anleitung wird ein End-to-End-Beispiel für die Erstellung eines neuen Datenfundierungsmoduls für ein Kundenservice-Ticketsystem beschrieben, dessen Daten (customers, tickets, ticketlogitem-Tabellen) aus einer PostgreSQL-Datenbank in ein BigQuery-Rohdatenset namens ticketing_data_raw repliziert wurden.

Wenn Sie ein benutzerdefiniertes Datenfundierungs-Modul erstellen, empfehlen wir die Verwendung eines dedizierten benutzerdefinierten Namespace, um das Lebenszyklusmanagement zu verbessern, indem Erweiterungen und Anpassungen von Cortex Framework-Artefakten getrennt werden.

Übersicht über das Beispielszenario

In dieser Anleitung werden wir:

  1. Einen dedizierten benutzerdefinierten Namespace ticketing erstellen, um die Datenfundierungs-Assets zu isolieren.
  2. Ein neues Datenfundierungsmodul mit dem Pfad ticketing.ticketing.foundations.ticketing_system definieren.
  3. Tabelleneinstellungen (table_settings.default.yaml) für die Tabellen customers, tickets und ticketlogitem konfigurieren.
  4. Metadaten-Annotationen auf Spalten- und Feldebene für jede Tabelle erstellen.
  5. Die Rohdatenquelle (ticketing_data_raw), das konforme Ziel-Dataset (data_foundation_ticketing) und das neue Fundierungsmodul in config/config.yaml registrieren.

Modulordner- und Dateistruktur

Alle physischen Dateien für Ihr neues Datenfundierungsmodul befinden sich in Ihrem benutzerdefinierten Namespace unter src/data_modules/. In der folgenden Tabelle und im Verzeichnisbaum wird beschrieben, wo die einzelnen Dateien platziert werden müssen:

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

Wichtig: Bevor Sie beginnen, prüfen Sie, ob die Quelltabelle, die Sie verarbeiten möchten, im Dataset der Rohdatenebene vorhanden ist.

Datei- oder Verzeichnispfad Zweck und Beschreibung
config/config.yaml Registriert den Namespace ticketing, die PostgreSQL-Rohdatenquelle, das BigQuery-Ziel-Dataset und die Instanz des Fundierungsmoduls.
src/data_modules/ticketing/ticketing/foundations/ticketing_system/manifest.yaml Deklariert die Modulmetadaten, den Anzeigenamen, die Kategorie (z.B. foundation), den Modultyp (z.B. generic) und die Generator-Builder-Klasse, die während der Kompilierung verwendet wird.
src/data_modules/ticketing/ticketing/foundations/ticketing_system/table_settings.default.yaml Konfiguriert, welche Quelltabelle aus ticketing_data_raw konform gemacht werden sollen, zusammen mit BigQuery-Optimierungs-Dataform-Tags, BigQuery-Labels, Partitionierungsdetails und Clusterdetails.
src/data_modules/ticketing/ticketing/foundations/ticketing_system/annotations/*.yaml Enthält umfangreiche YAML-Metadaten, die Tabellen- und Felddefinitionen beschreiben. Diese werden automatisch in die kompilierten Dataform-Definitionen zusammengeführt, sodass Beschreibungen in den BigQuery-Tabellenmetadaten erhalten bleiben.
src/data_modules/ticketing/ticketing/foundations/ticketing_system/builder.py Optional. Wenn Ihre Quelldatenbank während der Kompilierung eine benutzerdefinierte Datenbereinigung oder dialektspezifische SQL-Transformationen erfordert, können Sie hier eine Builder-Klasse auf Modulebene definieren.

Schritt 1: Namespace, Datenquelle und Ziel in config.yaml registrieren

Öffnen Sie vor dem Erstellen der physischen Dateien die Bereitstellungskonfigurationsdatei (config/config.yaml) und deklarieren Sie den benutzerdefinierten Namespace, das PostgreSQL-Rohdatenset und das Ziel-Dataset, in dem konforme Tabellen erstellt werden:

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

Schritt 2: Datenfundierungsmodul in config.yaml registrieren

Registrieren Sie im Abschnitt data.modules.foundations von config/config.yaml die neue Instanz des Datenfundierungsmoduls und verknüpfen Sie die Datenquelle (ticketing_data_raw) mit dem Datenziel (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"

Schritt 3: Modulmanifestdatei erstellen

Erstellen Sie die Manifestdatei, in der Sie die Modulmetadaten deklarieren: 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

Schritt 4: Tabelleneinstellungsdatei (table_settings.default.yaml) erstellen

Erstellen Sie die Standardkonfigurationsdatei für Tabellen: src/data_modules/ticketing/ticketing/foundations/ticketing_system/table_settings.default.yaml. In dieser Datei wird definiert, wie die replizierten PostgreSQL-Tabellen (customers, tickets, ticketlogitem) in BigQuery materialisiert, partitioniert und gruppiert werden:

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]

Schritt 5: Metadaten-Annotationen auf Feldebene erstellen

Damit Ihre konformen Tabellen in BigQuery eine klare Dokumentation enthalten, erstellen Sie für jede Tabelle eine YAML-Datei mit Annotationen in src/data_modules/ticketing/ticketing/foundations/ticketing_system/annotations/. Der Dateiname muss genau mit dem Namen der Quelltabelle übereinstimmen.

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"

Schritt 6 (optional): Benutzerdefinierten Fundierungs-Builder definieren

Wenn Ihre PostgreSQL-Datenfundierung während der Kompilierung Logik erfordert (z. B. automatische Datentypumwandlung, Zeitstempelkonvertierung oder Regeln zur Datenbereinigung für alle Tabellen), können Sie einen benutzerdefinierten Builder definieren, der auf dieses Modul beschränkt ist.

Erstellen Sie 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)

Überprüfung des neuen Fundierungsmoduls

So überprüfen und stellen Sie Ihr neu erstelltes Datenfundierungsmodul bereit:

  1. Führen Sie das Build- und Bereitstellungsskript von Cortex Framework aus: bash uv run cortex-build-and-deploy --config "config/config.yaml"
  2. Prüfen Sie, ob die Dataform-Kompilierung ohne Fehler abgeschlossen wurde und ob .sqlx-Skripts für customers, tickets und ticketlogitem generiert wurden.
  3. Führen Sie die Schritte nach der Bereitstellung aus, um die Dataform-Pipelineaktionen auszuführen und die konformen Datensätze in Ihrem data_foundation_ticketing Dataset in BigQuery zu überprüfen.

Informationen zum Überprüfen, ob das benutzerdefinierte Datenfundierungsmodul erfolgreich kompiliert und bereitgestellt wurde, finden Sie im Abschnitt „Überprüfung“ auf der Seite zur Erweiterbarkeit von Datenprodukten.