Ingerir dados do SQL Server

Saiba como ingerir dados do SQL Server no Azure Databricks usando o Lakeflow Connect.

O conector do SQL Server dá suporte ao Banco de Dados SQL do Azure, à Instância Gerenciada de SQL do Azure e aos bancos de dados SQL do Amazon RDS. Isso inclui o SQL Server em execução em VMs (máquinas virtuais) do Azure e no Amazon EC2. O conector também dá suporte ao SQL Server local usando a rede do Azure ExpressRoute e do AWS Direct Connect.

Requisitos

• Para criar um gateway de ingestão e um pipeline de ingestão, primeiro você deve atender aos seguintes requisitos:

  • Seu espaço de trabalho está habilitado para o Unity Catalog.

  • A computação sem servidor está habilitada para seu workspace. Consulte os requisitos de computação sem servidor.

  • Se você planeja criar uma conexão, você tem privilégios CREATE CONNECTION no metastore. Consulte Gerenciar privilégios no Catálogo do Unity.

    Se o conector for compatível com a criação de pipeline baseada na interface do usuário, você poderá criar a conexão e o pipeline ao mesmo tempo seguindo os passos desta página. No entanto, se você usar a criação de pipeline baseada em API, deverá criar a conexão no Catalog Explorer antes de concluir as etapas nesta página. Consulte Conectar-se às fontes de ingestão gerenciadas.

  • Se você planeja usar uma conexão existente: tem privilégios USE CONNECTION ou ALL PRIVILEGES sobre a conexão.

  • Você tem privilégios USE CATALOG no catálogo de destino.

  • Você tem privilégios USE SCHEMA, CREATE TABLE e CREATE VOLUME em um esquema existente ou privilégios CREATE SCHEMA no catálogo de destino.

  • Você tem acesso a uma instância primária do SQL Server. Os recursos de rastreamento de alterações e captura de dados alterados não contam com suporte em réplicas de leitura ou instâncias secundárias.

  • Permissões irrestritas para criar clusters ou uma política personalizada (somente API). Uma política personalizada para o gateway deve atender aos seguintes requisitos:

    • Família: Computação de Trabalho

    • Substituições da família de políticas:

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

    • O Databricks recomenda especificar os menores nós de trabalho possíveis para gateways de ingestão porque eles não afetam o desempenho do gateway. A política de computação a seguir permite que o Azure Databricks dimensione o gateway de ingestão para atender às necessidades de sua carga de trabalho. O requisito mínimo para o nó do driver é de 8 núcleos para permitir uma extração eficiente e de alto desempenho dos dados do seu banco de dados de origem.

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

    Para obter mais informações sobre políticas de cluster, consulte Selecionar uma política de computação.

• Para ingerir do SQL Server, primeiro você deve concluir as etapas em Configurar o Microsoft SQL Server para ingestão no Azure Databricks.

Criar um gateway e um pipeline de ingestão

Interface do usuário do Databricks

1. Na barra lateral do workspace do Azure Databricks, clique em Ingestão de Dados.

2. Na página Adicionar dados , em conectores do Databricks, clique em SQL Server.

3. Na página Connection do assistente de ingestão, selecione a conexão que armazena suas credenciais de acesso SQL Server. Se você tiver o privilégio CREATE CONNECTION no metastore, você pode clicar em Criar conexão para criar uma nova conexão com os detalhes de autenticação em Criar uma conexão SQL Server.

4. Clique em Próximo.

5. Na página Configuração de ingestão, insira um nome exclusivo para o pipeline de ingestão. Esse pipeline move os dados do local de preparação para o destino.

6. Selecione um catálogo e um esquema para o qual gravar logs de eventos. O log de eventos inclui auditorias, verificações de qualidade de dados, progresso do pipeline e erros. Se você tiver privilégios USE CATALOG e CREATE SCHEMA no catálogo, poderá clicar em Criar esquema no menu suspenso para criar um novo esquema.

7. (Opcional) Defina a atualização completa automática para todas as tabelas como Ativadas. Quando a atualização automática está ativada, o pipeline tenta corrigir automaticamente problemas como eventos de limpeza de log e determinados tipos de evolução do esquema atualizando totalmente a tabela afetada. Se o controle de histórico estiver habilitado, uma atualização completa apagará esse histórico.

8. Insira um nome exclusivo para o gateway de ingestão. O gateway é um pipeline que extrai as alterações da origem e as prepara para que o pipeline de ingestão seja carregado.

9. Selecione um catálogo e um esquema para o local de preparo. Um volume é criado nesse local para preparar os dados extraídos. Se você tiver privilégios USE CATALOG e CREATE SCHEMA no catálogo, poderá clicar em Criar esquema no menu suspenso para criar um novo esquema.

10. Clique em Criar pipeline e continuar.

11. Na página Origem , selecione as tabelas a serem ingeridas. Se você selecionar tabelas específicas, poderá definir as configurações da tabela:

    a. (Opcional) Na guia Configurações, especifique um nome de destino para cada tabela ingerida. Isso é útil para diferenciar entre tabelas de destino quando você ingerir um objeto no mesmo esquema várias vezes. Consulte Nome de uma tabela de destino.

    a. (Opcional) Altere a configuração de controle de histórico padrão. Consulte Habilitar acompanhamento de histórico (SCD tipo 2).

12. Clique em Avançar e, em seguida, clique em Salvar e continuar.

13. Na página Destino , selecione um catálogo e um esquema para carregar dados. Se você tiver privilégios USE CATALOG e CREATE SCHEMA no catálogo, poderá clicar em Criar esquema no menu suspenso para criar um novo esquema.

14. Clique em Salvar e continuar.

15. Na página de configuração do Banco de Dados , clique em Validar para confirmar se sua origem está configurada corretamente para ingestão do Azure Databricks. Todas as configurações ausentes são retornadas. Para obter as etapas a serem resolvidas, clique em Concluir configuração. Em seguida, clique em Próximo. Como alternativa, clique em Ignorar validação.

16. (Opcional) Na página Agendas e notificações , clique no Criar agendamento. Defina a frequência para atualizar as tabelas de destino.

17. (Opcional) Clique Adicione uma notificação para definir notificações por email para êxito ou falha na operação do pipeline e clique em Salvar e executar pipeline.

Pacotes de Automação Declarativa

Antes de ingerir usando Pacotes de Automação Declarativa, você deve ter acesso a uma conexão existente. Para obter instruções, consulte Criar uma conexão SQL Server.

O catálogo e o esquema de preparo podem ser os mesmos que o catálogo e o esquema de destino. O catálogo de preparação não pode ser um catálogo desconhecido. Especifique o local de preparação na seção gateway_definition do arquivo YAML do pipeline do pacote.

O gateway de ingestão extrai os dados de instantâneo e alteração do banco de dados de origem e os armazena no volume de preparação do Catálogo do Unity. É preciso operacionalizar o gateway como pipeline contínuo. Isso ajuda a acomodar todas as políticas de retenção de log de alterações que você tem no banco de dados de origem.

O pipeline de ingestão aplica os dados de alteração e instantâneo do volume de preparo nas tabelas de streaming de destino.

Os pacotes podem conter definições YAML de trabalhos e tarefas, são gerenciados usando a CLI do Databricks e podem ser compartilhados e executados em diferentes workspaces de destino (como desenvolvimento, preparo e produção). Para obter mais informações, consulte o que são pacotes de automação declarativa?.

1. Crie um pacote usando a CLI do Databricks:

   databricks bundle init
   

2. Adicione o pipeline e a configuração do trabalho ao pacote. Consulte Exemplos para obter um exemplo completo com todas as opções disponíveis.

3. Implante o pipeline usando a CLI do Databricks:

   databricks bundle deploy
   

Bloco de anotações do Databricks

Atualize a célula Configuration no notebook a seguir com a conexão de origem, catálogo de destino, esquema de destino e tabelas a serem ingeridas da origem.

Obter laptop

Terraform

Você pode usar o Terraform para implantar e gerenciar pipelines de ingestão do SQL Server. Para obter uma estrutura de exemplo completa, incluindo configurações do Terraform para criar gateways e pipelines de ingestão, consulte o repositório de exemplos do Terraform do Lakeflow Connect no GitHub.

Verificar a ingestão de dados bem-sucedida

A exibição de lista na página de detalhes do pipeline mostra o número de registros processados quando os dados são ingeridos. Esses números são atualizados automaticamente.

As colunas Upserted records e Deleted records não são mostradas por padrão. Você pode habilitá-las clicando no botão de configuração de colunas e selecionando-as.

Exemplos

Use esses exemplos para configurar o pipeline.

Configuração do pipeline

Pacotes de Automação Declarativa

O pacote a seguir define um pipeline de gateway, um pipeline de ingestão e um trabalho agendado. As opções comentadas mostram todas as configurações disponíveis. Atualize as seções variables e targets com suas informações de origem e destino.

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>'

Bloco de anotações do Databricks

O seguinte é uma seção Configuration de exemplo da especificação de um pipeline:

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

Padrões comuns

Para configurações avançadas de pipeline, consulte padrões comuns para pipelines de ingestão gerenciada.

Próximas Etapas 

Inicie, agende e defina alertas no seu fluxo de trabalho. Consulte Tarefas Comuns de Manutenção de Pipeline.

Recursos adicionais

• Perguntas frequentes sobre o conector do SQL Server
• Limitações do conector do SQL Server
• Referência do conector do SQL Server
• Solucionar problemas de ingestão do SQL Server