Mata in data från SQL Server

Lär dig hur du matar in data från SQL Server till Azure Databricks med Lakeflow Connect.

SQL Server-anslutningsappen stöder Sql Database-, Azure SQL Managed Instance- och Amazon RDS SQL-databaser. Detta inkluderar SQL Server som körs på virtuella Azure-datorer (VM) och Amazon EC2. Anslutningsappen har också stöd för SQL Server lokalt med hjälp av Azure ExpressRoute- och AWS Direct Connect-nätverk.

Requirements

  • Om du vill skapa en inmatningsgateway och en inmatningspipeline måste du först uppfylla följande krav:

    • Arbetsytan är aktiverad för Unity Catalog.

    • Serverlös beräkning är aktiverad för din arbetsyta. Se Krav för serverlös beräkning.

    • Om du planerar att skapa en anslutning: Du har CREATE CONNECTION behörigheter i metastore. Se avsnitt Hantera privilegier i Unity Catalog.

      Om anslutningsappen stöder UI-baserad pipelineredigering kan du skapa anslutningen och pipelinen samtidigt genom att slutföra stegen på den här sidan. Men om du använder API-baserad pipelineredigering måste du skapa anslutningen i Katalogutforskaren innan du slutför stegen på den här sidan. Se Anslut till hanterade inmatningskällor.

    • Om du planerar att använda en befintlig anslutning: Du har USE CONNECTION behörighet eller ALL PRIVILEGES för anslutningen.

    • Du har USE CATALOG behörigheter i målkatalogen.

    • Du har USE SCHEMA, CREATE TABLE, och CREATE VOLUME behörigheter för ett befintligt schema eller CREATE SCHEMA behörigheter för målkatalogen.

    • Du har åtkomst till en primär SQL Server-instans. Funktioner för ändringsspårning och ändringsdatainsamling stöds inte på läsrepliker eller sekundära instanser.

    • Obegränsade behörigheter för att skapa kluster eller en anpassad princip (endast API). En anpassad princip för gatewayen måste uppfylla följande krav:

      • Familj: Jobbberäkning

      • Policypaketets åsidosättningar

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

      • Databricks rekommenderar att du anger de minsta möjliga arbetsnoderna för inmatningsgatewayer eftersom de inte påverkar gatewayprestanda. Med följande beräkningsprincip kan Azure Databricks skala inmatningsgatewayen så att den uppfyller arbetsbelastningens behov. Minimikravet är 8 kärnor för att möjliggöra effektiv och högpresterande dataextrahering från källdatabasen.

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

      Mer information om klusterprinciper finns i Välj en beräkningsprincip.

  • Om du vill mata in från SQL Server måste du först slutföra stegen i Konfigurera Microsoft SQL Server för inmatning till Azure Databricks.

Skapa en gateway och en inmatningspipeline

Databricks-användargränssnitt

  1. I sidofältet på Azure Databricks-arbetsytan klickar du på Datainmatning.

  2. På sidan Lägg till data under Databricks-anslutningsappar klickar du på SQL Server.

  3. På sidan Anslutning i inmatningsguiden väljer du den anslutning som lagrar autentiseringsuppgifter för SQL Server-åtkomst från Konfigurera Microsoft SQL Server för inmatning till Azure Databricks. Om du har behörigheten CREATE CONNECTION för metaarkivet kan du klicka på Plus-ikonen. Skapa en anslutning för att skapa en ny anslutning med autentiseringsinformationen i SQL Server.

  4. Klicka på Nästa.

  5. På installationssidan för inmatning anger du ett unikt namn för inmatningspipelinen. Den här pipelinen flyttar data från mellanlagringsplatsen till målet.

  6. Välj en katalog och ett schema att skriva händelseloggar till. Händelseloggen innehåller granskningsloggar, datakvalitetskontroller, pipeline-förlopp och fel. Om du har USE CATALOG och CREATE SCHEMA behörigheter i katalogen kan du klicka på Plus-ikonen. Skapa ett schema i den nedrullningsbara menyn för att skapa ett nytt schema.

  7. (Valfritt) Ange Automatisk fullständig uppdatering för alla tabeller till På. När automatisk uppdatering är på försöker pipelinen automatiskt åtgärda problem som loggrensningshändelser och vissa typer av schemautveckling genom att helt uppdatera den påverkade tabellen. Om historikspårning är aktiverat raderar en fullständig uppdatering den historiken.

  8. Ange ett unikt namn för inmatningsgatewayen. Gatewayen är en pipeline som extraherar ändringar från källan och förbereder dem för att inmatningspipelinen ska kunna ladda in dem.

  9. Välj en katalog och ett schema för mellanlagringsplatsen. En volym skapas på den här platsen för att mellanlagra extraherade data. Om du har USE CATALOG och CREATE SCHEMA behörigheter i katalogen kan du klicka på Plus-ikonen. Skapa ett schema i den nedrullningsbara menyn för att skapa ett nytt schema.

  10. Klicka på Skapa pipeline och fortsätt.

  11. På sidan Källa väljer du de tabeller som ska matas in. Om du väljer specifika tabeller kan du konfigurera tabellinställningar:

    a. (Valfritt) På fliken Inställningar anger du ett målnamn för varje inmatad tabell. Detta är användbart för att skilja mellan måltabeller när du matar in ett objekt i samma schema flera gånger. Se Namnge en måltabell.

    a. (Valfritt) Ändra standardinställningen för spårning av historik . Se Aktivera historikspårning (SCD-typ 2).

  12. Klicka på Nästa och sedan på Spara och fortsätt.

  13. På sidan Mål väljer du en katalog och ett schema att läsa in data i. Om du har USE CATALOG och CREATE SCHEMA behörigheter i katalogen kan du klicka på Plus-ikonen. Skapa ett schema i den nedrullningsbara menyn för att skapa ett nytt schema.

  14. Klicka på Spara och fortsätt.

  15. På sidan Databaskonfiguration klickar du på Verifiera för att bekräfta att källan är korrekt konfigurerad för Azure Databricks-inmatning. Eventuella saknade konfigurationer returneras. Klicka på Slutför konfiguration för att lösa det. Klicka sedan på Nästa. Du kan också klicka på Hoppa över validering.

  16. (Valfritt) På sidan Scheman och meddelanden klickar du på plusikonen. Skapa schema. Ange frekvensen för att uppdatera måltabellerna.

  17. (Valfritt) Klicka på Plus-ikonen. Lägg till meddelande för att ange e-postaviseringar för lyckade eller misslyckade pipelineåtgärder och klicka sedan på Spara och kör pipeline.

Deklarativa automationspaket

Innan du importerar med deklarativa automationspaket måste du ha åtkomst till en befintlig anslutning. Anvisningar finns i Ansluta till hanterade inmatningskällor.

Mellanlagringskatalogen och schemat kan vara samma som målkatalogen och schemat. Mellanlagringskatalogen får inte vara en extern katalog. Ange mellanlagringsplatsen i gateway_definition-avsnittet i YAML-filen för din paketpipeline.

Inmatningsgatewayen extraherar ögonblicksdata och förändringsdata från källdatabasen och lagrar dem i Unity Catalogs mellanlagringsvolym. Gatewayen måste köras som en kontinuerlig pipeline. Detta hjälper dig att hantera eventuella kvarhållningsprinciper för ändringsloggar som du har i källdatabasen.

Inmatningspipelinen tillämpar ögonblicksbilden och ändrar data från mellanlagringsvolymen till målströmningstabeller.

Paket kan innehålla YAML-definitioner av jobb och uppgifter, hanteras med Databricks CLI och kan delas och köras på olika målarbetsytor (till exempel utveckling, mellanlagring och produktion). Mer information finns i Vad är deklarativa automatiseringspaket?.

  1. Skapa ett paket med Databricks CLI:

    databricks bundle init

  2. Lägg till två nya resursfiler i paketet:

    • En pipelinedefinitionsfil (till exempel resources/sqlserver_pipeline.yml). Se pipeline.ingestion_definition och exempel.
    • En jobbdefinitionsfil som styr datainmatningsfrekvensen (till exempel resources/sqlserver_job.yml).

  3. Distribuera pipelinen med Databricks CLI:

    databricks bundle deploy

Databricks-anteckningsbok

Uppdatera cellen Configuration i följande notebook-fil med källanslutningen, målkatalogen, målschemat och tabellerna som ska matas in från källan.

Hämta anteckningsbok

Terraform

Du kan använda Terraform för att distribuera och hantera SQL Server-inmatningspipelines. Ett komplett exempelramverk, inklusive Terraform-konfigurationer för att skapa gatewayer och inmatningspipelines, finns i lagringsplatsen Lakeflow Connect Terraform-exempel på GitHub.

Verifiera lyckad datainmatning

Listvyn på sidan med pipelineinformation visar antalet poster som bearbetas när data matas in. Dessa siffror uppdateras automatiskt.

Verifiera replikering

Kolumnerna Upserted records och Deleted records visas inte som standard. Du kan aktivera dem genom att klicka på ikonen för konfiguration av kolumner och välja dem.

Exempel

Använd de här exemplen för att konfigurera din pipeline.

Konfiguration för pipeline

Deklarativa automationspaket

Följande pipelinedefinitionsfil:

variables:
  # Common variables used multiple places in the DAB definition.
  gateway_name:
    default: sqlserver-gateway
  dest_catalog:
    default: main
  dest_schema:
    default: ingest-destination-schema

resources:
  pipelines:
    gateway:
      name: ${var.gateway_name}
      gateway_definition:
        connection_name: <sqlserver-connection>
        gateway_storage_catalog: main
        gateway_storage_schema: ${var.dest_schema}
        gateway_storage_name: ${var.gateway_name}
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}

    pipeline_sqlserver:
      name: sqlserver-ingestion-pipeline
      ingestion_definition:
        ingestion_gateway_id: ${resources.pipelines.gateway.id}
        objects:
          # Modify this with your tables!
          - table:
              # Ingest the table test.ingestion_demo_lineitem to dest_catalog.dest_schema.ingestion_demo_line_item.
              source_catalog: test
              source_schema: ingestion_demo
              source_table: lineitem
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
          - schema:
              # Ingest all tables in the test.ingestion_whole_schema schema to dest_catalog.dest_schema. The destination
              # table name will be the same as it is on the source.
              source_catalog: test
              source_schema: ingestion_whole_schema
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
      catalog: ${var.dest_catalog}
      schema: ${var.dest_schema}

Databricks-anteckningsbok

Följande är ett exempelavsnitt Configuration i en pipelinespecifikation:

# 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"])

Paketjobbsdefinitionsfil

Följande är ett exempel på en jobbdefinitionsfil för användning med deklarativa Automation-paket. Jobbet körs varje dag, precis 24 timmar efter den senaste körningen.

resources:
  jobs:
    sqlserver_dab_job:
      name: sqlserver_dab_job

      trigger:
        periodic:
          interval: 1
          unit: DAYS

      email_notifications:
        on_failure:
          - <email-address>

      tasks:
        - task_key: refresh_pipeline
          pipeline_task:
            pipeline_id: ${resources.pipelines.pipeline_sqlserver.id}

Vanliga mönster

För avancerade pipelinekonfigurationer, se Vanliga mönster för hanterade inmatningspipelines.

Nästa steg

Starta, schemalägga och ange aviseringar för din pipeline. Se Vanliga pipelineunderhållsuppgifter.

Ytterligare resurser

  • Last updated on