Pular para o conteúdo principal
Última atualização em

Ingerir dados do Apache Kafka

info

Beta

Este recurso está em Beta. Os administradores do espaço de trabalho podem controlar o acesso a esse recurso na página Pré-visualizações . Consulte Gerenciar prévias do Databricks.

Esta página mostra como criar um pipeline de ingestão gerenciado do Kafka usando o Databricks LakeFlow Connect.

Requisitos​

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

    • Seu workspace deve estar habilitado para o Unity Catalog.

    • O compute serverless deve ser habilitado para seu workspace. Consulte Requisitos de computação serverless.

    • Para criar uma nova conexão, é preciso ter os privilégios CREATE CONNECTION no metastore. Consulte Gerenciar privilégios no Unity Catalog.

      Se o conector suportar a criação de pipelines baseada na IU, um administrador poderá criar a conexão e o pipeline simultaneamente, ao concluir os passos nesta página. No entanto, se os usuários que criam pipelines usam a autoria de pipeline baseada em API ou são usuários não administradores, um administrador deve primeiro criar a conexão no Catalog Explorer. Consulte Conectar-se a fontes de ingestão gerenciadas.

    • Para usar uma conexão existente: é preciso ter os privilégios USE CONNECTION ou ALL PRIVILEGES no objeto de conexão.

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

    • É necessário ter os privilégios USE SCHEMA e CREATE TABLE em um esquema existente ou os privilégios CREATE SCHEMA no catálogo de destino.

  • Para ingerir do Kafka, primeiro, conclua os passos em Conectar ao Apache Kafka para ingestão gerenciada.

Criar pipeline de ingestão​

Cada tópico Kafka é ingerido em uma tabela de transmissão. Para obter uma lista de dados compatíveis e limitações, consulte Dados compatíveis.

nota

A criação de pipelines baseada em interface de usuário não é compatível com o conector Kafka na versão Beta. Use Pacotes de Automação Declarativa ou um Notebook Databricks para criar seu pipeline.

Use Pacotes de Automação Declarativa para gerenciar pipelines Kafka como código. Os pacotes podem conter definições YAML de Jobs 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 novo pacote utilizando a CLI do Databricks:

    Bash
    databricks bundle init

  2. Adicione um arquivo de definição de pipeline ao pacote (por exemplo, resources/kafka_pipeline.yml). Veja pipeline.definição_de_ingestão e Exemplos.

  3. Implante o pacote usando a CLI do Databricks:

    Bash
    databricks bundle deploy

Exemplos​

Use estes exemplos para configurar seu pipeline.

Pipeline mínima — key e valor binários brutos​

Este exemplo ingere um ou mais tópicos Kafka com colunas de key e de valor retidas como BINARY:

YAML
variables:
  connection_name:
    default: my-kafka-connection
  dest_catalog:
    default: main
  dest_schema:
    default: kafka_ingest
resources:
  pipelines:
    kafka_pipeline:
      name: kafka-ingestion-pipeline
      serverless: true
      continuous: true
      channel: PREVIEW
      catalog: ${var.dest_catalog}
      target: ${var.dest_schema}
      ingestion_definition:
        connection_name: ${var.connection_name}
        objects:
          - table:
              source_table: N/A
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              destination_table: user_events
              connector_options:
                kafka_options:
                  topics: [user-events, power-user-events]

Pipeline com transformadores — valor JSON e key de string​

Este exemplo desserializa as chaves de mensagem como STRING e os valores como JSON com a evolução do esquema habilitada:

YAML
variables:
  connection_name:
    default: my-kafka-connection
  dest_catalog:
    default: main
  dest_schema:
    default: kafka_ingest
resources:
  pipelines:
    kafka_pipeline:
      name: kafka-ingestion-pipeline
      serverless: true
      continuous: true
      channel: PREVIEW
      catalog: ${var.dest_catalog}
      target: ${var.dest_schema}
      ingestion_definition:
        connection_name: ${var.connection_name}
        objects:
          - table:
              source_table: N/A
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
              destination_table: user_events
              connector_options:
                kafka_options:
                  topics:
                    - user-events
                  starting_offset: latest
                  key_transformer:
                    format: STRING
                  value_transformer:
                    format: JSON
                    json_options:
                      schema_evolution_mode: rescue

Padrões comuns​

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

Passos seguintes​

Comece e defina alertas no seu pipeline. Veja Tarefas comuns de manutenção de pipelines.

Recursos adicionais​

Nesta página