Gegevens opnemen uit SQL Server

Meer informatie over het opnemen van gegevens van SQL Server in Azure Databricks met behulp van Lakeflow Connect.

De SQL Server-connector ondersteunt Azure SQL Database-, Azure SQL Managed Instance- en Amazon RDS SQL-databases. Dit omvat SQL Server die wordt uitgevoerd op virtuele Azure-machines (VM's) en Amazon EC2. De connector ondersteunt ook on-premises SQL Server met behulp van Azure ExpressRoute en AWS Direct Connect-netwerken.

Requirements

• Als u een opnamegateway en een opnamepijplijn wilt maken, moet u eerst aan de volgende vereisten voldoen:

  • Uw werkruimte is ingeschakeld voor Unity Catalog.

  • Serverloze rekenkracht is ingeschakeld voor uw werkruimte. Zie serverloze rekenvereisten.

  • Als u een verbinding wilt maken: u hebt CREATE CONNECTION bevoegdheden voor de metastore. Zie Beheer van bevoegdheden in Unity Catalog.

    Als uw connector het ontwerpen van pijplijnen op basis van de gebruikersinterface ondersteunt, kunt u de verbinding en de pijplijn tegelijkertijd maken door de stappen op deze pagina uit te voeren. Als u echter op API gebaseerde pijplijncreatie gebruikt, moet u de verbinding maken in Catalog Explorer voordat u de stappen op deze pagina voltooit. Zie Verbinding maken met beheerde opnamebronnen.

  • Als u van plan bent een bestaande verbinding te gebruiken: u hebt USE CONNECTION bevoegdheden of ALL PRIVILEGES voor de verbinding.

  • U hebt USE CATALOG rechten voor de doelcatalogus.

  • U hebt USE SCHEMA, CREATE TABLEen CREATE VOLUME bevoegdheden voor een bestaand schema of CREATE SCHEMA bevoegdheden voor de doelcatalogus.

  • U hebt toegang tot een primair SQL Server-exemplaar. Functies voor het bijhouden en vastleggen van wijzigingen worden niet ondersteund op leesreplika's of secundaire instanties.

  • Onbeperkte machtigingen voor het maken van clusters of een aangepast beleid (alleen API). Een aangepast beleid voor de gateway moet voldoen aan de volgende vereisten:

    • Gezin: Taakberekening

    • Uitzonderingen van beleidsfamilies

      {
        "cluster_type": {
          "type": "fixed",
          "value": "dlt"
        },
        "num_workers": {
          "type": "unlimited",
          "defaultValue": 1,
          "isOptional": true
        },
        "runtime_engine": {
          "type": "fixed",
          "value": "STANDARD",
          "hidden": true
        }
      }
      

    • Databricks raadt aan om de kleinste mogelijke werkknooppunten voor opnamegateways op te geven, omdat deze geen invloed hebben op de gatewayprestaties. Met het volgende rekenbeleid kan Azure Databricks de opnamegateway schalen om te voldoen aan de behoeften van uw workload. De minimale vereiste voor de drivernode is 8 cores om efficiënte en effectieve data-extractie uit je brondatabase mogelijk te maken.

      {
        "driver_node_type_id": {
          "type": "fixed",
          "value": "Standard_E64d_v4"
        },
        "node_type_id": {
          "type": "fixed",
          "value": "Standard_F4s"
        }
      }
      

    Zie Een rekenbeleid selecteren voor meer informatie over clusterbeleid.

• Als u wilt opnemen vanuit SQL Server, moet u eerst de stappen in Microsoft SQL Server configureren voor opname in Azure Databricks uitvoeren.

Een gateway en een opnamepijplijn maken

Databricks-gebruikersinterface

1. Klik in de zijbalk van de Azure Databricks-werkruimte op Gegevensopname.

2. Klik op de pagina Gegevens toevoegen onder Databricks-connectors op SQL Server.

3. Selecteer op de pagina Connection van de opnamewizard de verbinding waarmee uw SQL Server toegangsreferenties worden opgeslagen. Als u de bevoegdheid CREATE CONNECTION hebt voor de metastore, U kunt op Verbinding maken klikken om een nieuwe verbinding te maken met de verificatiedetails in Maak een SQL Server-verbinding.

4. Klik op Volgende.

5. Voer op de pagina Invoerconfiguratie een unieke naam in voor de invoerpijplijn. Met deze pijplijn worden gegevens van de faseringslocatie naar het doel verplaatst.

6. Selecteer een catalogus en een schema om gebeurtenislogboeken naar te schrijven. Het gebeurtenislogboek bevat auditlogboeken, controles van gegevenskwaliteit, voortgang van pijplijn en fouten. Als u de bevoegdheden USE CATALOG en CREATE SCHEMA op de catalogus hebt, kunt u op Schema maken in de vervolgkeuzelijst om een nieuw schema te maken.

7. (Optioneel) Automatisch volledig vernieuwen instellen voor alle tabellen op Aan. Wanneer automatisch vernieuwen is ingeschakeld, probeert de pijplijn automatisch problemen op te lossen, zoals gebeurtenissen voor het opschonen van logboeken en bepaalde typen schemaontwikkeling door de betrokken tabel volledig te vernieuwen. Als het bijhouden van geschiedenis is ingeschakeld, wordt die geschiedenis gewist door een volledige vernieuwing.

8. Voer een unieke naam in voor de opnamegateway. De gateway is een pijplijn die wijzigingen uit de bron extraheert en klaarmaakt voor de ingestiepijplijn om te laden.

9. Selecteer een catalogus en een schema voor de faseringslocatie. Op deze locatie wordt een volume gemaakt om geëxtraheerde gegevens te organiseren. Als u de bevoegdheden USE CATALOG en CREATE SCHEMA op de catalogus hebt, kunt u op Schema maken in de vervolgkeuzelijst om een nieuw schema te maken.

10. Klik op Pijplijn maken en ga door.

11. Selecteer op de pagina Bron de tabellen die u wilt opnemen. Als u specifieke tabellen selecteert, kunt u tabelinstellingen configureren:

    a. (Optioneel) Geef op het tabblad Instellingen een doelnaam op voor elke opgenomen tabel. Dit is handig om onderscheid te maken tussen doeltabellen wanneer u een object meerdere keren in hetzelfde schema opneemt. Zie Plaats een naam voor een doeltabel.

    a. (Optioneel) Wijzig de standaardinstelling voor het bijhouden van geschiedenis . Zie Geschiedenis bijhouden inschakelen (SCD-type 2).

12. Klik op Volgende en klik vervolgens op Opslaan en doorgaan.

13. Selecteer op de doelpagina een catalogus en een schema om gegevens in te laden. Als u de bevoegdheden USE CATALOG en CREATE SCHEMA op de catalogus hebt, kunt u op Schema maken in de vervolgkeuzelijst om een nieuw schema te maken.

14. Klik op Opslaan en doorgaan.

15. Klik op de pagina Database-instelling op Valideren om te bevestigen dat uw bron juist is geconfigureerd voor opname van Azure Databricks. Ontbrekende configuraties worden geretourneerd. Klik op Configuratie voltooien voor stappen die u wilt oplossen. Klik vervolgens op Volgende. U kunt ook op Validatie overslaan klikken.

16. (Optioneel) Klik op de pagina Planningen en meldingen op Maak een planning. Stel de frequentie in om de doeltabellen te vernieuwen.

17. (Optioneel) Klik op Voeg een melding toe om e-mailmeldingen in te stellen voor geslaagde of mislukte pijplijnen en klik vervolgens op Opslaan en pijplijn uitvoeren.

Declaratieve automatiseringsbundels

Voordat u gegevens importeert met behulp van Declaratieve Automatiseringsbundels, moet u toegang hebben tot een bestaande verbinding. Zie Maak een SQL Server-verbinding voor instructies.

De faseringscatalogus en het schema kunnen hetzelfde zijn als de doelcatalogus en het schema. De faseringscatalogus kan geen buitenlandse catalogus zijn. Geef de faseringslocatie op in de gateway_definition sectie van het YAML-bestand voor de bundelpijplijn.

De opnamegateway extraheert momentopnamen en wijzigingsgegevens uit de brondatabase en slaat deze op in het tussenvolume van de Unity Catalog. Je moet de gateway uitvoeren als een continue pipeline. Dit helpt bij het instellen van bewaarbeleid voor wijzigingenlogboeken dat u in de brondatabase hebt.

De opnamepijplijn past de momentopname toe en brengt wijzigingsgegevens van het faseringsvolume aan in streamingdoeltabellen.

Bundels kunnen YAML-definities van taken en taken bevatten, worden beheerd met behulp van de Databricks CLI en kunnen worden gedeeld en uitgevoerd in verschillende doelwerkruimten (zoals ontwikkeling, fasering en productie). Zie Wat zijn declaratieve Automation-bundels? voor meer informatie.

1. Maak een bundel met behulp van de Databricks CLI:

   databricks bundle init
   

2. Voeg uw pijplijn- en taakconfiguratie toe aan de bundel. Zie Voorbeelden voor een volledig voorbeeld met alle beschikbare opties.

3. Implementeer de pijplijn met behulp van de Databricks CLI:

   databricks bundle deploy
   

Databricks Notebook

Werk de Configuration cel in het volgende notebook bij met de bronverbinding, doelcatalogus, doelschema en tabellen die moeten worden opgenomen uit de bron.

Notitieblok ophalen

Terraform

U kunt Terraform gebruiken om SQL Server-opnamepijplijnen te implementeren en beheren. Voor een volledig voorbeeldframework, waaronder Terraform-configuraties voor het maken van gateways en opnamepijplijnen, raadpleegt u de opslagplaats met voorbeelden van Lakeflow Connect Terraform op GitHub.

Geslaagde gegevensopname verifiëren

In de lijstweergave op de pagina met pijplijndetails ziet u het aantal records dat wordt verwerkt wanneer gegevens worden opgenomen. Deze nummers worden automatisch vernieuwd.

De Upserted records kolommen en Deleted records kolommen worden niet standaard weergegeven. U kunt deze inschakelen door op het te klikken en ze te selecteren.

Examples

Gebruik deze voorbeelden om uw pijplijn te configureren.

Pijplijnconfiguratie

Declaratieve automatiseringsbundels

De volgende bundel definieert een gateway-pijplijn, een innamepijplijn en een geplande taak. Uitgecommentarieerde opties tonen alle beschikbare configuratieopties. Werk de variables en targets secties bij met de bron- en doelgegevens.

bundle:
  name: lakeflow-connect-sqlserver

# Variables parameterize the bundle for different environments and sources.
# Set values here, override per-target, or pass with: databricks bundle deploy -var="key=value"
variables:
  # The name of the Unity Catalog connection to your SQL Server instance.
  # This connection must already exist and be of type SQLSERVER.
  connection_name:
    description: 'Unity Catalog connection name for the SQL Server source'
  # The SQL Server database name to ingest from.
  # In Lakeflow Connect, this maps to source_catalog in the table/schema spec.
  source_database:
    description: 'SQL Server database name (maps to source_catalog in table specs)'
  # The SQL Server schema to ingest from (for example, "dbo", "sales").
  source_schema:
    description: 'SQL Server schema name to ingest from'
  # The Unity Catalog catalog where ingested Delta tables are created.
  dest_catalog:
    description: 'Destination Unity Catalog catalog for ingested tables'
  # The Unity Catalog schema where ingested Delta tables are created.
  dest_schema:
    description: 'Destination Unity Catalog schema for ingested tables'
  # The Unity Catalog catalog for the gateway's internal staging volume.
  # Can be the same as dest_catalog. Must not be a foreign catalog.
  staging_catalog:
    description: 'Catalog for gateway staging volume'
  # The Unity Catalog schema for the gateway's internal staging volume.
  staging_schema:
    description: 'Schema for gateway staging volume'

resources:
  pipelines:
    # --- Gateway pipeline ---
    # Extracts change data from SQL Server and stages it in a Unity Catalog
    # volume. Must run continuously to capture changes before change logs are
    # truncated in the source database.
    gw_pipeline:
      name: 'lfc-sqlserver-gateway-${bundle.target}'
      # Gateway pipelines must be continuous.
      continuous: true
      # "CURRENT" (stable) or "PREVIEW" (early access).
      channel: 'CURRENT'
      # (Optional) Associate with a budget policy for cost tracking.
      # budget_policy_id: "<policy-uuid>"
      # The gateway runs on classic compute. Cluster settings are managed
      # automatically. You can optionally customize the cluster:
      # clusters:
      #   - label: "default"
      #     autoscale:
      #       min_workers: 1
      #       max_workers: 4
      #     # node_type_id: "i3.xlarge"
      #     # Restrict the cluster to an approved cluster policy.
      #     # policy_id: "<cluster-policy-id>"
      catalog: ${var.staging_catalog}
      schema: ${var.staging_schema}
      gateway_definition:
        # (Required) Unity Catalog connection name (type SQLSERVER).
        connection_name: ${var.connection_name}
        # (Required) Catalog and schema for the staging volume.
        gateway_storage_catalog: ${var.staging_catalog}
        gateway_storage_schema: ${var.staging_schema}
        # (Optional) Custom staging volume name. If not set, the system
        # auto-generates: __databricks_ingestion_gateway_staging_data-<pipeline_id>
        # gateway_storage_name: "my_custom_staging_volume"

    # --- Ingestion pipeline ---
    # Reads staged data from the gateway and applies it to Delta tables.
    mi_pipeline:
      name: 'lfc-sqlserver-ingestion-${bundle.target}'
      # Continuous mode is not supported for the ingestion pipeline.
      # Use a scheduled job to trigger runs.
      continuous: false
      channel: 'CURRENT'
      # (Optional) Associate with a budget policy for cost tracking.
      # budget_policy_id: "<policy-uuid>"
      # The ingestion pipeline runs on serverless compute only.
      serverless: true
      # (Optional) Development mode for faster iteration (no retries).
      # development: true
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}
      # (Optional) Email notifications for pipeline events.
      # notifications:
      #   - email_recipients:
      #       - "team@example.com"
      #     alerts:
      #       - "on-update-failure"
      #       - "on-update-fatal-failure"
      #       - "on-flow-failure"
      # (Optional) Run as a service principal for production.
      # run_as:
      #   service_principal_name: "my-service-principal"
      ingestion_definition:
        # (Required) References the gateway pipeline. The connection is
        # inherited from the gateway. Do not specify connection_name here.
        ingestion_gateway_id: ${resources.pipelines.gw_pipeline.id}
        # Pipeline-level table configuration defaults. These apply to all
        # tables unless overridden at the schema or table level.
        table_configuration:
          # SCD Type: How changes are applied to destination tables.
          #   SCD_TYPE_1: Overwrites rows with latest values (default).
          #   SCD_TYPE_2: Preserves history with __START_AT/__END_AT columns.
          #               Requires CDC on source. CT does not support SCD_TYPE_2.
          scd_type: 'SCD_TYPE_1'
          # (Optional) Auto full refresh policy. Triggers a snapshot when the
          # pipeline detects issues resolvable by re-reading all source data
          # (for example, CT/CDC retention window expired).
          # auto_full_refresh_policy:
          #   enabled: true
          #   min_interval_hours: 24
        # (Optional) Schedule automatic full refreshes.
        # full_refresh_window:
        #   start_hour: 2
        #   days_of_week:
        #     - "SUNDAY"
        #   time_zone_id: "America/Los_Angeles"
        objects:
          # Option 1: Schema-level ingestion. Ingests all tables from a source
          # schema. New tables added to the schema are picked up automatically.
          - schema:
              source_catalog: ${var.source_database}
              source_schema: ${var.source_schema}
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              # (Optional) Override table_configuration for this schema.
              # table_configuration:
              #   scd_type: "SCD_TYPE_2"
          # Option 2: Table-level ingestion. Provides granular control.
          # Replace or combine with the schema-level spec.
          # - table:
          #     source_catalog: ${var.source_database}
          #     source_schema: ${var.source_schema}
          #     source_table: "customers"
          #     destination_catalog: ${var.dest_catalog}
          #     destination_schema: ${var.dest_schema}
          #     # (Optional) Rename the table at the destination.
          #     # destination_table: "customers_v2"
          #     table_configuration:
          #       scd_type: "SCD_TYPE_1"
          #       # Include only specific columns (mutually exclusive with exclude_columns).
          #       # include_columns:
          #       #   - "customer_id"
          #       #   - "first_name"
          #       #   - "email"
          #       # Exclude specific columns. All other columns are included.
          #       # exclude_columns:
          #       #   - "internal_notes"
          #       # Override the primary key used for change detection.
          #       # primary_keys:
          #       #   - "customer_id"
          #       # Logical ordering columns for change resolution.
          #       # sequence_by:
          #       #   - "updated_at"
          #       # Auto full refresh for this table.
          #       # auto_full_refresh_policy:
          #       #   enabled: true
          #       #   min_interval_hours: 48
      # (Optional) Grant additional users or groups access.
      # permissions:
      #   - user_name: "analyst@example.com"
      #     level: "CAN_VIEW"
      #   - group_name: "data-engineers"
      #     level: "CAN_RUN"

  # --- Scheduled job ---
  # Triggers the ingestion pipeline on a schedule.
  jobs:
    mi_schedule:
      name: 'lfc-sqlserver-ingestion-schedule-${bundle.target}'
      # Quartz cron syntax: "seconds minutes hours day month day-of-week"
      # Examples: "0 0 * * * ?" (hourly), "0 0 */4 * * ?" (every 4 hours)
      schedule:
        quartz_cron_expression: '0 */30 * * * ?'
        timezone_id: 'UTC'
      tasks:
        - task_key: 'run_ingestion'
          pipeline_task:
            pipeline_id: ${resources.pipelines.mi_pipeline.id}
      # email_notifications:
      #   on_failure:
      #     - "team@example.com"

# Deploy to different workspaces with: databricks bundle deploy -t <target>
targets:
  dev:
    default: true
    workspace:
      host: https://<workspace-url>.cloud.databricks.com
    variables:
      connection_name: '<sqlserver-connection>'
      source_database: '<database-name>'
      source_schema: 'dbo'
      dest_catalog: '<dest-catalog>'
      dest_schema: '<dest-schema>'
      staging_catalog: '<staging-catalog>'
      staging_schema: '<staging-schema>'

Databricks Notebook

Hier volgt een voorbeeldsectie Configuration van een pijplijnspecificatie:

# The name of the UC connection with the credentials to access the source database
connection_name = "my_connection"

# The name of the UC catalog and schema to store the replicated tables
target_catalog_name = "main"
target_schema_name = "lakeflow_sqlserver_connector_cdc"

# The name of the UC catalog and schema to store the staging volume with intermediate
# CDC and snapshot data. Use the destination catalog/schema by default.
stg_catalog_name = target_catalog_name
stg_schema_name = target_schema_name

# The name of the Gateway pipeline to create
gateway_pipeline_name = "cdc_gateway"

# The name of the Ingestion pipeline to create
ingestion_pipeline_name = "cdc_ingestion"

# Construct the full list of tables to replicate.
# IMPORTANT: The letter case of catalog, schema, and table names must match exactly
# the case used in the source database system tables.
tables_to_replicate = replicate_full_db_schema("MY_DB", ["MY_DB_SCHEMA"])
# Append tables from additional schemas as needed:
#  + replicate_tables_from_db_schema("MY_DB", "MY_SCHEMA_2", ["table3", "table4"])

Algemene patronen

Zie Algemene patronen voor beheerde opnamepijplijnen voor geavanceerde pijplijnconfiguraties.

Volgende stappen 

Start, plan en stel waarschuwingen in voor uw pijplijn. Zie Algemene onderhoudstaken voor pijplijnen.

Aanvullende bronnen

• Veelgestelde vragen over SQL Server-connector
• Beperkingen voor SQL Server-connector
• Naslaginformatie voor SQL Server-connector
• Problemen met SQL Server-inname oplossen