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

Utilize destinos em pipelines

Use a API sink do Lakeflow Spark Declarative Pipelines com fluxos para gravar registros transformados por um pipeline em um coletor de dados externo. Os destinos de dados externos incluem tabelas gerenciadas e externas do Unity Catalog, e serviços de transmissão de eventos como Apache Kafka ou Azure Event Hubs. Você também pode usar coletores de dados para gravar em fontes de dados personalizadas escrevendo código Python para essa fonte de dados.

Para uma visão geral dos conceitos de destino e quando usá-los, consulte Destinos em Lakeflow Spark Declarative Pipelines.

nota

Sink fluxo de trabalho​

À medida que os dados de eventos são ingeridos de uma fonte de transmissão em seu pipeline, você processa e refina esses dados em transformações em seu pipeline. Em seguida, você 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. Utilize um fluxo de acréscimo ou um fluxo de atualização para gravar os registros preparados no destino.

Crie uma pia​

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

  • Destinos de tabelas Delta (incluindo gerenciamento Unity Catalog e tabelas externas)
  • Apache Kafka sinks
  • Destinos do Azure Event Hubs
  • Sumidouros personalizados escritos em Python, usando a fonte de dados personalizada Python

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

Para criar um coletor Delta por caminho de 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 caminho de catálogo e esquema totalmente qualificado:

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

Para obter mais detalhes sobre como usar a função create_sink , consulte a referência da API de destino.

Após a criação do seu ponto de coleta, você pode começar a transmitir os registros processados para ele.

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

Com o seu coletor criado, o próximo passo é gravar os registros processados nele, especificando-o como o destino para os registros emitidos por um fluxo de acréscimo. Você faz isso especificando seu sink 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 coletor Kafka Spark suporta. Consulte Configurar o gravador de transmissão estruturada Kafka.
  • Para o Azure Event Hubs, use o formato kafka e especifique o nome do Event Hubs, 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 de Event Hubs estruturado Spark que usa a interface Kafka . Consulte Autenticação.

Abaixo estão exemplos de como configurar fluxos para gravar em coletores 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 default flows e append flows.

Limitações​

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

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

  • Somente append_flow e update_flow podem ser usados para gravar em coletores. Outros fluxos, como create_auto_cdc_flow, não são compatíveis, e não é possível usar um destino em uma definição de dataset de pipeline. Por exemplo, o seguinte não é aceito:

    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 o Unity Catalog gerenciar tabelas externas, o nome da tabela deve estar no formato <catalog>.<schema>.<table>. Para o Hive metastore, deve estar no formato <schema>.<table>.

  • A execução de uma atualização de refresh completa não limpa os dados de resultados de compute computados anteriormente nos sinks. Isso significa que todos os dados reprocessados são anexados ao coletor e os dados existentes não são alterados.

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

  • O controle de saída sem servidor suporta apenas conectores de destino Kafka e Delta Lake . Veja O que é controle de saída serverless ?

Recursos adicionais​

Nesta página