Erstellen einer integrierten CDC-Pipeline für SQL Server

Eine integrierte CDC-Pipeline übernimmt Änderungsdaten aus SQL Server über eine einzige Pipeline in Azure Databricks. Im Gegensatz zur herkömmlichen Gateway-basierten Architektur, die ein separates Erfassungs-Gateway und eine separate Erfassungspipeline erfordert, führt eine integrierte CDC-Pipeline sowohl die Extraktions- als auch die Anwendungsphase in einer einzigen Pipeline-Aktualisierung aus.

Wann der integrierte CDC-Connector verwendet werden sollte

In der folgenden Tabelle werden integrierte CDC-Pipelines mit der standardmäßigen gatewaybasierten Architektur verglichen:

Informationen zum Einrichten der Quelldatenbank finden Sie unter Configure-Microsoft SQL Server für die Aufnahme in Azure Databricks. Die gleiche Quellkonfiguration gilt für beide Architekturen.

So funktioniert eine integrierte CDC-Pipeline

Jede Pipeline-Aktualisierung führt zwei Phasen nacheinander aus:

1. Extraktion. Die Pipeline stellt mithilfe der Unity Catalog-Verbindung eine Verbindung mit der Quelldatenbank her. Bei der ersten Ausführung oder einer vollständigen Aktualisierung wird eine anfängliche Momentaufnahme erfasst. Bei nachfolgenden Ausführungen werden inkrementelle Änderungen (Einfügungen, Aktualisierungen und Löschungen) mithilfe des integrierten Änderungsnachverfolgungsmechanismus der Datenbank erfasst. Die Pipeline schreibt extrahierte Daten in ein Unity Catalog-Stagingvolume.
2. Anwendung: Die Pipeline liest aus dem Stagingvolume und wendet Änderungen auf die Zielstreamingtabellen in Unity-Katalog an. Zusammenführungsvorgänge verwenden die konfigurierten Primärschlüssel und den SCD-Typ. Die Pipeline garantiert Exactly-once-Semantik.

Jede Pipeline-Aktualisierung extrahiert Änderungen und stoppt dann automatisch, sobald sie den Stand der Quelle erreicht hat, begrenzt durch eine maximale Laufzeit. Ausführliche Informationen finden Sie unter "Intelligentes Schließen" für integrierte CDC-Pipelines. Zum Aufnehmen von Daten auf wiederkehrender Basis planen Sie die Pipeline mit einem Lakeflow Jobs-Vorgang .

Requirements

• Ihr Arbeitsbereich ist für Unity Catalog aktiviert.

• Wenn Sie eine Verbindung erstellen möchten: Sie verfügen über CREATE CONNECTION Berechtigungen für den Metastore. Weitere Informationen finden Sie unter Verwalten von Berechtigungen in Unity Catalog.

  Wenn Ihr Connector die benutzeroberflächenbasierte Pipelineerstellung unterstützt, können Sie die Verbindung und die Pipeline gleichzeitig erstellen, indem Sie die Schritte auf dieser Seite ausführen. Wenn Sie jedoch apibasierte Pipelineerstellung verwenden, müssen Sie die Verbindung im Katalog-Explorer erstellen, bevor Sie die Schritte auf dieser Seite ausführen. Siehe Herstellen einer Verbindung mit verwalteten Aufnahmequellen.

• Wenn Sie beabsichtigen, eine vorhandene Verbindung zu verwenden: Sie verfügen über USE CONNECTION Berechtigungen oder ALL PRIVILEGES für die Verbindung.

• Sie verfügen über USE CATALOG-Berechtigungen für den Zielkatalog.

• Sie verfügen über USE SCHEMA, CREATE TABLE und CREATE VOLUME Berechtigungen für ein vorhandenes Schema oder über CREATE SCHEMA Berechtigungen für den Zielkatalog.

• Für Ihren Arbeitsbereich muss die integrierte CDC-Connectorfunktion aktiviert sein. Wenden Sie sich an Ihr Azure Databricks Kontoteam.
• Sie haben Zugriff auf die instanz primary SQL Server. Der integrierte CDC-Connector unterstützt keine Lesereplikate, Standbyinstanzen oder sekundäre Instanzen.
• Sie haben die Einrichtung der SQL Server-Quelle abgeschlossen. Siehe Microsoft SQL Server für die Aufnahme in Azure Databricks konfigurieren.
• Sie verfügen über die folgenden Berechtigungen:
  • CREATE CONNECTION auf dem Metastore (wenn eine neue Unity-Katalogverbindung erstellt wird) oder USE CONNECTION auf einer vorhandenen Verbindung.
  • USE CATALOG im Zielkatalog.
  • USE SCHEMA und CREATE TABLE im Zielschema.
  • CREATE VOLUME auf dem Zielschema oder auf dem in data_staging_options angegebenen Schema. Ein Staging-Volume ist auch dann erforderlich, wenn data_staging_options nicht festgelegt ist, da die Pipeline im Zielschema automatisch eines erstellt.

Computeanforderungen

Eine integrierte CDC-Pipeline wird auf klassischem oder serverlosen Compute ausgeführt:

• Klassischer Compute: Die klassische Compute-Ebene wird im VPC- oder VNet-Netzwerk Ihres Azure Databricks-Arbeitsbereichs ausgeführt und muss in der Lage sein, Ihre SQL Server-Instanz über das Netzwerk zu erreichen. Jeder Netzwerkpfad, der es der Compute-Ebene ermöglicht, die Datenbank zu erreichen, wird unterstützt, einschließlich VPC- oder VNet-Peering, öffentlicher Endpunkte und, für lokale SQL Server-Instanzen, AWS Direct Connect, Azure ExpressRoute oder VPN.
• Serverlose Berechnung: Konfigurieren Sie serverlose Netzwerkkonnektivität zwischen Azure Databricks serverlosen Compute und Ihrer Quelldatenbank. Lokale Quellen erfordern einen Netzwerkpfad über den konfigurierten serverlosen Ausgang (z. B. ein Transitgateway oder ein gepeertes VNet mit ExpressRoute oder VPN).

Für klassisches Compute können Sie uneingeschränkte Berechtigungen zum Erstellen von Clustern oder eine benutzerdefinierte Clusterrichtlinie verwenden, wobei cluster_type auf dlt festgelegt und runtime_engine auf STANDARD festgelegt ist und für eine effiziente Extraktion mindestens 8 Kerne empfohlen werden.

Erstellen Sie eine Unity Catalog-Verbindung mit SQL Server

Erstellen Sie eine Unity-Katalogverbindung mit SQL Server, bevor Sie eine Pipeline erstellen. Siehe Create a SQL Server connection.

Erstellen einer integrierten CDC-Pipeline

Erstellen Sie integrierte CDC-Pipelines mit der API, der Databricks CLI, Notizbüchern oder deklarativen Automatisierungspaketen. Das Erstellen von Benutzeroberflächen ist noch nicht verfügbar.

Deklarative Automatisierungspakete

Definieren Sie die Pipeline-Ressource in einer Bundle-Datei (z. B. resources/integrated_cdc_pipeline.yml):

variables:
  pipeline_name:
    description: 'Name for the integrated CDC pipeline'
  connection_name:
    description: 'Unity Catalog connection name'
  dest_catalog:
    description: 'Destination catalog for ingested data'
  dest_schema:
    description: 'Destination schema for ingested data'

resources:
  pipelines:
    integrated_cdc_pipeline:
      name: ${var.pipeline_name}
      channel: PREVIEW
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}
      ingestion_definition:
        connection_name: ${var.connection_name}
        connector_type: CDC
        objects:
          - table:
              source_catalog: 'my_database'
              source_schema: 'dbo'
              source_table: 'customers'
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              destination_table: 'customers'
              table_configuration:
                scd_type: 'SCD_TYPE_1'

Um die Pipeline in einem Zeitplan auszuführen, definieren Sie einen Auftrag (z. B resources/integrated_cdc_job.yml. ), der die Pipeline auslöst. Da jede Extraktionsphase mindestens 10 Minuten dauert, ist ein Intervall von 60 Minuten oder länger ein guter Ausgangspunkt:

resources:
  jobs:
    integrated_cdc_job:
      name: '${var.pipeline_name}-job'
      tasks:
        - task_key: 'cdc_ingestion'
          pipeline_task:
            pipeline_id: ${resources.pipelines.integrated_cdc_pipeline.id}
      schedule:
        quartz_cron_expression: '0 0 * * * ?'
        timezone_id: 'UTC'

Stellen Sie das Bundle mit der Databricks CLI bereit:

databricks bundle deploy
databricks bundle run integrated_cdc_job

Weitere Informationen finden Sie unter Was sind deklarative Automatisierungs-Bundles?.

Databricks-Notizbuch

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.pipelines import (
    ConnectorType,
    IngestionConfig,
    IngestionPipelineDefinition,
    TableSpec,
)

w = WorkspaceClient()

pipeline = w.pipelines.create(
    name="<pipeline-name>",
    channel="PREVIEW",
    catalog="<destination-catalog>",
    schema="<destination-schema>",
    ingestion_definition=IngestionPipelineDefinition(
        connection_name="<unity-catalog-connection-name>",
        connector_type=ConnectorType.CDC,
        objects=[
            IngestionConfig(
                table=TableSpec(
                    source_catalog="<source-database>",
                    source_schema="<source-schema>",
                    source_table="<source-table>",
                    destination_catalog="<destination-catalog>",
                    destination_schema="<destination-schema>",
                )
            )
        ],
    ),
)

print(f"Pipeline created: {pipeline.pipeline_id}")

Databricks-Befehlszeilenschnittstelle

databricks pipelines create --json '{
  "name": "<pipeline-name>",
  "channel": "PREVIEW",
  "catalog": "<destination-catalog>",
  "schema": "<destination-schema>",
  "ingestion_definition": {
    "connection_name": "<unity-catalog-connection-name>",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "<source-database>",
          "source_schema": "<source-schema>",
          "source_table": "<source-table>"
        }
      }
    ]
  }
}'

REST API

Im folgenden Beispiel werden zwei Tabellen aus einer SQL Server Datenbank repliziert. In der Tabelle customers wird SCD Typ 1 verwendet, und die tabelle orders verwendet SCD Typ 2 (für die SQL Server CDC auf der Quelle erforderlich ist). Beide erben das Ziel main.ingestionder obersten Ebene. Im Beispiel wird serverless weggelassen; standardmäßig wird false (klassische Rechenleistung) verwendet. Fügen Sie stattdessen "serverless": true hinzu, um die Ausführung auf serverlosen Rechenressourcen durchzuführen.

POST /api/2.0/pipelines

{
  "name": "my-integrated-cdc-pipeline",
  "channel": "PREVIEW",
  "catalog": "main",
  "schema": "ingestion",
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "customers",
          "table_configuration": {
            "scd_type": "SCD_TYPE_1"
          }
        }
      },
      {
        "table": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "source_table": "orders",
          "table_configuration": {
            "scd_type": "SCD_TYPE_2"
          }
        }
      }
    ],
    "data_staging_options": {
      "catalog_name": "main",
      "schema_name": "ingestion_staging"
    }
  }
}

Verwenden Sie schema ein Objekt anstelle einzelner table Objekte, um jede Tabelle in einem Quellschema zu replizieren. Die Pipeline überspringt Tabellen, für die in der Quelle weder CDC noch die Änderungsnachverfolgung aktiviert ist.

POST /api/2.0/pipelines

{
  "name": "my-integrated-cdc-schema-pipeline",
  "channel": "PREVIEW",
  "catalog": "main",
  "schema": "ingestion",
  "ingestion_definition": {
    "connection_name": "my-sqlserver-connection",
    "connector_type": "CDC",
    "objects": [
      {
        "schema": {
          "source_catalog": "my_database",
          "source_schema": "dbo",
          "destination_catalog": "main",
          "destination_schema": "ingestion"
        }
      }
    ]
  }
}

So starten Sie ein Pipeline-Update:

POST /api/2.0/pipelines/<pipeline-id>/updates

{
  "full_refresh": false
}

Wiederkehrende Updates planen

Standardmäßig laufen integrierte CDC-Pipelines im Trigger-Modus. Für die always-on-Ausführung siehe Eine integrierte CDC-Pipeline im kontinuierlichen Modus ausführen. Um Daten nach einem wiederkehrenden Zeitplan zu erfassen, erstellen Sie eine Lakeflow-Einzelvorgangs-Aufgabe, die die Pipeline ausführt. Die Aktualisierungsdauer variiert mit der Menge der Änderungsdaten, die die Quelle hat, und ein großer Backlog wird möglicherweise nicht in einem einzigen Update abgeschlossen (siehe intelligentes Schließen für integrierte CDC-Pipelines). Planen Sie die Ausführung von Pipelines häufig genug, damit nachfolgende Updates aufholen können. Ein Ausgangspunkt von 60 Minuten eignet sich gut für die meisten Workloads. Wenn ein Trigger ausgelöst wird, während noch ein vorheriges Update ausgeführt wird, wird das neue Update in die Warteschlange gestellt.

Konfigurationsreferenz

Pipelineparameter

Tabellendefinition

Tabellenkonfiguration

Informationen zu SQL Server Datentypzuordnungen finden Sie unter SQL Server Connectorreferenz. Integrierte CDC-Pipelines unterstützen die automatische Erweiterung von Datentypen: Wenn der Datentyp einer Quellspalte erweitert wird (z. B. von INT zu BIGINT), passt sich die Zieltabelle automatisch an.

Überwachen der Pipeline

Nachdem Sie eine integrierte CDC-Pipeline erstellt und gestartet haben, überwachen Sie den Status mithilfe der folgenden Schritte:

• Azure Databricks UI. Öffnen Sie die Pipeline im Abschnitt Pipelines, um den Aktualisierungsstatus, die Erfassungsmetriken pro Tabelle und die Herkunft anzuzeigen.

• REST-API.

  GET /api/2.0/pipelines/<pipeline-id>
  

• Ereignis-API.

  GET /api/2.0/pipelines/<pipeline-id>/events
  

Das erste Update der Pipeline erstellt einen vollständigen Snapshot aller ausgewählten Tabellen, was länger dauern kann als inkrementelle Updates. Bei großen Tabellen erfordert die anfängliche Momentaufnahme möglicherweise mehrere geplante Aktualisierungen. Jedes nachfolgende Update macht dort weiter, wo das vorherige aufgehört hat.

So überprüfen Sie die Erfassung:

-- Check row counts in the destination table
SELECT COUNT(*) FROM <destination_catalog>.<destination_schema>.<destination_table>;

-- View recent changes (SCD Type 2 tables)
SELECT * FROM <destination_catalog>.<destination_schema>.<destination_table>
ORDER BY __START_AT DESC
LIMIT 10;

Weitere Informationen zur vollständigen Aktualisierung und zum Verhalten der automatischen vollständigen Aktualisierung finden Sie unter Zieltabellen vollständig aktualisieren.

Integrierte CDC-Pipelines verfügen standardmäßig über eine vertikale Automatische Skalierung. Wenn ein Pipeline-Update aufgrund von Speichermangel fehlschlägt, wird beim nächsten Update automatisch ein größerer Driver bereitgestellt. Verwenden Sie eine benutzerdefinierte Clusterrichtlinie, um dieses Verhalten außer Kraft zu setzen.

Einschränkungen

• Beta. Der integrierte CDC-Connector erfordert die Aktivierung auf Arbeitsbereichsebene. Wenden Sie sich an Ihr Azure Databricks Kontoteam.
• Standardmäßig ausgelöst. Standardmäßig laufen integrierte CDC-Pipelines im ausgelösten Modus; planen Sie sie mit einer Lakeflow-Jobs-Aufgabe. Der kontinuierliche Modus ist in der Beta verfügbar. Siehe Ausführen einer integrierten CDC-Pipeline im kontinuierlichen Modus.
• Erstellung nur über die API. Die Pipelineerstellung ist über die REST-API, die Databricks CLI, Notizbücher und deklarative Automatisierungspakete verfügbar. Die Ui-Erstellung wird noch nicht unterstützt.
• Kanal muss sein PREVIEW. Pipelinespezifikationen müssen "channel": "PREVIEW" enthalten.
• Verbindungs- und Verbindertyp sind unveränderlich. connection_name und connector_type kann nicht geändert werden, nachdem die Pipeline erstellt wurde. Um die Quelle zu ändern, erstellen Sie eine neue Pipeline.
• Empfohlener Höchstwert von 300 Tabellen pro Pipeline.
• Nur primäre Instanzen. Der integrierte CDC-Connector unterstützt keine Lesereplikate, Standbyinstanzen oder sekundäre Instanzen.
• Tabellen ohne Primärschlüssel. Die Pipeline behandelt alle Nicht-LOB-Spalten als einen zusammengesetzten Schlüssel. Doppelte Zeilen können auf eine einzelne Zeile reduziert werden, es sei denn, Sie aktivieren SCD Type 2.
• Die erste Momentaufnahme kann sich über mehrere Updates erstrecken. Bei großen Tabellen wird die anfängliche Momentaufnahme möglicherweise nicht in einem einzigen Update abgeschlossen. Nachfolgende geplante Updates werden fortgesetzt, wo das vorherige Update unterbrochen wurde.
• Updatelaufzeit wird automatisch verwaltet: Intelligentes Schließen bestimmt, wann jedes Update beendet wird. Ein Update wird abgeschlossen, nachdem es den Stand der Quelle erreicht hat, wobei es an eine maximale Laufzeit gebunden ist. Siehe Intelligentes Schließen für integrierte CDC-Pipelines. Sie können die Mindest- oder Maximallaufzeit nicht konfigurieren. Ein großer Änderungsbacklog kann mehrere Updates umfassen. Nachfolgende geplante Updates werden fortgesetzt, wo das vorherige Update unterbrochen wurde.
• Für die Protokolllöschung ist eine vollständige Aktualisierung erforderlich. Wenn SQL Server Änderungsnachverfolgungsprotokolle oder CDC-Protokolle löscht, bevor die Pipeline sie verarbeitet, führen Sie für die betroffenen Tabellen eine vollständige Aktualisierung durch. Die Pipeline erkennt diesen Zustand und meldet einen Fehler im Ereignisprotokoll.

Problembehandlung

Wenn ein Problem auftritt, das hier nicht behandelt wird:

1. Überprüfen Sie das Ereignisprotokoll der Pipeline in der Azure Databricks-Benutzeroberfläche oder über GET /api/2.0/pipelines/<pipeline-id>/events.
2. Testen Sie die Unity-Katalogverbindung aus dem Katalog-Explorer, um zu bestätigen, dass die Quelle erreichbar ist.
3. Vergewissern Sie sich, dass die Änderungsnachverfolgung oder das CDC in der Quelldatenbank und -tabellen aktiviert ist.
4. Stellen Sie sicher, dass der Datenbankbenutzer über die SQL Server Berechtigungen verfügt, die in Microsoft SQL Server-Datenbank Benutzeranforderungen aufgeführt sind.
5. Überprüfen Sie, ob Ihre Pipelinespezifikation "channel": "PREVIEW" enthält.

Weitere Ressourcen

• Microsoft SQL Server Erfassungskonnektor
• Konfigurieren von Microsoft SQL Server für die Aufnahme in Azure Databricks
• SQL Server-Connectorreferenz
• Fehlerbehebung bei der Datenverarbeitung in SQL Server
• Vollständige Aktualisierung von Zieltabellen