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

Use sinks para transmitir registros para serviço externo com pipeline declarativo LakeFlow Spark

info

Visualização

A API LakeFlow Spark Declarative pipeline sink está em Pré-visualização Pública.

Este artigo descreve a API LakeFlow Spark Declarative pipeline sink e como usá-la com fluxos para gravar registros transformados por um pipeline em um coletor de dados externo. Os destinos de dados externos incluem Unity Catalog , tabelas externas e serviços de transmissão de eventos como Apache Kafka ou Azure Event Hubs.

nota

A API sink está disponível apenas para Python.

O que são pias?​

Os sumidouros são destinos para os fluxos em um pipeline. Por default, os fluxos pipeline emitem dados para uma tabela de transmissão ou para view materializada. Ambas são tabelas Delta Databricks . Os sinks são um destino alternativo que você usa para gravar dados transformados em destinos como serviços de transmissão de eventos, como Apache Kafka ou Azure Event Hubs, e tabelas externas gerenciadas pelo Unity Catalog. Ao usar sinks, você agora tem mais opções para persistir a saída do seu pipeline.

Quando devo usar as pias?​

A Databricks recomenda o uso de sinks se você precisar:

  • Crie um caso de uso operacional, como detecção de fraudes, análise de tempo real e recomendações de clientes. Os casos de uso operacional normalmente leem dados de um barramento de mensagens, como um tópico do Apache Kafka, e então processam dados com baixa latência e gravam os registros processados de volta em um barramento de mensagens. Essa abordagem permite que você alcance menor latência ao não gravar ou ler dados do armazenamento cloud .
  • Grave os dados transformados dos seus fluxos em tabelas gerenciadas por uma instância Delta externa, incluindo Unity Catalog e tabelas externas.
  • Execute extração reversa, transformação, carregamento (ETL) em coletores externos ao Databricks, como tópicos Apache Kafka . Essa abordagem permite que você ofereça suporte eficaz a casos de uso em que os dados precisam ser lidos ou usados fora das tabelas Unity Catalog ou de outro armazenamento gerenciado Databricks .
  • É necessário gravar em um formato de dados que não é suportado diretamente pelo Databricks. A fonte de dados personalizada Python permite que você crie um coletor que grava em qualquer fonte de dados usando código Python personalizado. Consulte Fonte de dados personalizadaPySpark.

Como usar as pias?​

À medida que os dados de eventos são ingeridos de uma fonte de transmissão em seu pipeline, você processa e refina essa funcionalidade pipeline de uso de dados e, em seguida, usa o processamento de fluxo de anexação para transmitir os registros de dados transformados para um destino. Você cria este coletor usando a função create_sink() . Para obter mais detalhes sobre a função create_sink , consulte a referência da API de destino.

Se você possui um pipeline que cria ou processa seus dados de eventos de transmissão e prepara registros de dados para gravação, então você está pronto para usar um coletor (sink).

A implementação de uma pia consiste em duas etapas:

  1. Crie a pia.
  2. Use um fluxo de acréscimo para gravar os registros preparados no coletor.

Crie uma pia​

Databricks suporta vários tipos de coletores de destino nos quais você grava seus registros processados a partir dos seus dados de transmissão:

  • Pias de tabela Delta (incluindo tabelas externas e gerenciais Unity Catalog )
  • Apache Kafka afunda
  • Coletores do Azure Event Hubs
  • Pias personalizadas escritas em Python, usando a fonte de dados personalizada Python

abaixo estão exemplos de configurações para os coletores Delta, Kafka e Azure Event Hubs, e fonte de dados personalizada Python :

Para criar um coletor Delta pelo caminho do arquivo:

Python
dp.create_sink(
  name = "delta_sink",
  format = "delta",
  options = {"path": "/Volumes/catalog_name/schema_name/volume_name/path/to/data"}
)

Para criar um coletor Delta por nome de tabela usando um catálogo totalmente qualificado e um caminho de esquema:

Python
dp.create_sink(
  name = "delta_sink",
  format = "delta",
  options = { "tableName": "catalog_name.schema_name.table_name" }
)

Para obter mais detalhes sobre o uso da função create_sink , consulte a referência da API do sink.

Depois que seu coletor for criado, você pode começar a transmitir registros processados para ele.

Escreva em um coletor com um fluxo de acréscimo​

Com seu coletor criado, o próximo passo é gravar registros processados nele, especificando-o como o destino para os registros de saída de um fluxo de acréscimo. Você faz isso especificando seu coletor como o valor target no decorador append_flow .

  • Para gerenciar Unity Catalog e tabelas externas, use o formato delta e especifique o caminho ou o nome da tabela nas opções. Seu pipeline deve ser configurado para usar Unity Catalog.
  • Para tópicos Apache Kafka , use o formato kafka e especifique o nome do tópico, as informações de conexão e as informações de autenticação nas opções. Essas são as mesmas opções que um dissipador Spark transmissão estructurada Kafka suporta. Consulte Configurar o gravador de transmissão estruturada Kafka.
  • Para os Hubs de Eventos Azure , use o formato kafka e especifique o nome dos Hubs de Eventos, as informações de conexão e as informações de autenticação nas opções. Essas são as mesmas opções suportadas em um coletor Spark Transmissão Estruturada Event Hubs que usa a interface Kafka . Consulte Autenticação da entidade de serviço com o Microsoft Entra ID e os Hubs de Eventos do Azure.

Abaixo estão exemplos de como configurar fluxos para gravar em Delta, Kafka e Azure Event Hubs com registros processados pelo seu pipeline.

Python
@dp.append_flow(name = "delta_sink_flow", target="delta_sink")
def delta_sink_flow():
  return(
  spark.readStream.table("spark_referrers")
  .selectExpr("current_page_id", "referrer", "current_page_title", "click_count")
)

Para obter mais detalhes sobre o decorador append_flow , consulte Usando vários fluxos para gravar em um único destino.

Limitações​

  • Somente a API Python é suportada. SQL não é suportado.

  • Somente consultas de transmissão são suportadas. Consultas de lotes não são suportadas.

  • Somente append_flow pode ser usado para escrever em sinks. Outros fluxos, como create_auto_cdc_flow, não são suportados e você não pode usar um coletor em uma definição de dataset de pipeline declarativo do LakeFlow Spark . Por exemplo, o seguinte não é suportado:

    Python
    @table("from_sink_table")
    def fromSink():
      return read_stream("my_sink")

  • Para sumidouros Delta, o nome da tabela deve ser totalmente qualificado. Especificamente, para tabelas externas Unity Catalog , o nome da tabela deve estar no formato <catalog>.<schema>.<table>. Para o Hive metastore, ele deve estar no formato <schema>.<table>.

  • Executar uma refresh completa não limpa os dados de resultados computados anteriormente nos coletores. Isso significa que todos os dados reprocessados serão anexados ao coletor e os dados existentes não serão alterados.

  • As expectativas de pipeline não são suportadas.

recurso​