Compartilhar via

Streaming e ingestão incremental

  • Artigo

O Azure Databricks usa o Apache Spark Structured Streaming para oferecer suporte a vários produtos associados a cargas de trabalho de ingestão, incluindo:

  • Carregador Automático
  • COPY INTO
  • Pipelines do Delta Live Tables
  • Exibições materializadas e tabelas de streaming no Databricks SQL

Este artigo discute algumas das diferenças entre a semântica de streaming e processamento em lotes incremental e fornece uma visão geral de alto nível da configuração de cargas de trabalho de ingestão para a semântica desejada no Databricks.

Qual é a diferença entre streaming e ingestão em lote incremental?

As possíveis configurações de fluxo de trabalho de ingestão variam de processamento quase em tempo real a processamento em lote incremental pouco frequente. Ambos os padrões usam o Apache Spark Structured Streaming para alimentar o processamento incremental, mas têm semântica diferente. Para simplificar, este artigo refere-se à ingestão quase em tempo real como ingestão de streaming e ao processamento incremental mais pouco frequente como ingestão em lote incremental.

Ingestão de streaming

O streaming, no contexto de ingestão de dados e atualizações de tabela, refere-se ao processamento de dados quase em tempo real em que o Azure Databricks ingere registros da origem para o coletor em microlotes usando a infraestrutura sempre ativa. Uma carga de trabalho de streaming ingere continuamente atualizações de fontes de dados configuradas, a menos que uma falha que interrompa a ingestão ocorra.

Ingestão em lote incremental

Ingestão em lote incremental refere-se a um padrão em que todos os novos registros são processados de uma fonte de dados em um trabalho de curta duração. A ingestão em lote incremental muitas vezes ocorre de acordo com um agendamento, mas também pode ser acionada manualmente ou com base na chegada do arquivo.

A ingestão incremental em lote difere da ingestão em lote no sentido em que detecta automaticamente novos registros na fonte de dados e ignora os registros que já foram assimilados.

Ingestão com trabalhos

Os trabalhos do Databricks permitem orquestrar fluxos de trabalho e agendar tarefas que incluem notebooks, bibliotecas, pipelines do Delta Live Tables e consultas SQL do Databricks.

Observação

Você pode usar todos os tipos de computação e tipos de tarefa do Azure Databricks para configurar a ingestão em lote incremental. A ingestão de streaming só é aceita na produção em trabalhos clássicos, computação e Delta Live Tables.

Os trabalhos têm dois modos principais de operação:

  • Os trabalhos contínuos são repetidos automaticamente se encontrarem uma falha. Esse modo destina-se à ingestão de streaming.
  • Os trabalhos disparados executam tarefas quando disparados. Os gatilhos incluem:
    • Gatilhos baseados em tempo que executam trabalhos em um agendamento especificado.
    • Gatilhos baseados em arquivo que executam trabalhos quando os arquivos chegam a um local especificado.
    • Outros gatilhos, como chamadas à API REST, execução de comandos da CLI do Azure Databricks ou cliques no botão Executar agora na interface do usuário do workspace.

Para cargas de trabalho em lote incrementais, configure seus trabalhos usando o modo de gatilho AvailableNow, da seguinte maneira:

Python

(df.writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(availableNow=True)
  .toTable("table_name")
)

Scala

import org.apache.spark.sql.streaming.Trigger

df.writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(Trigger.AvailableNow)
  .toTable("table_name")

Para cargas de trabalho de streaming, o intervalo de disparo padrão é processingTime ="500ms". O exemplo a seguir mostra como processar um microlote a cada 5 segundos:

Python

(df.writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(processingTime="5 seconds")
  .toTable("table_name")
)

Scala

import org.apache.spark.sql.streaming.Trigger

df.writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(Trigger.ProcessingTime, "5 seconds")
  .toTable("table_name")

Importante

Os trabalhos sem servidor não são compatíveis com Scala, modo contínuo ou intervalos de gatilho baseados em tempo para fluxo estruturado. Use trabalhos clássicos se precisar de semântica de ingestão quase em tempo real.

Ingestão com Delta Live Tables

Semelhante aos trabalhos, os pipelines do Delta Live Tables podem ser executados no modo de disparo ou contínuo. Para semântica de streaming quase em tempo real com tabelas de streaming, use o modo contínuo.

Use tabelas de streaming para configurar o streaming ou a ingestão incremental em lote do armazenamento de objetos na nuvem, Apache Kafka, Amazon Kinesis, Google Pub/Sub ou Apache Pulsar.

O LakeFlow Connect usa Delta Live Tables para configurar pipelines de ingestão de sistemas conectados. Consulte LakeFlow Connect.

As exibições materializadas garantem a semântica da operação equivalente às cargas de trabalho em lote, mas podem otimizar muitas operações para calcular os resultados de forma incremental. Confira Operações de atualização para exibições materializadas.

Ingestão com o Databricks SQL

Você pode usar tabelas de streaming para configurar a ingestão incremental em lote do armazenamento de objetos na nuvem, Apache Kafka, Amazon Kinesis, Google Pub/Sub ou Apache Pulsar.

Você pode usar exibições materializadas para configurar o processamento em lote incremental de fontes que são totalmente reproduzíveis para um conjunto especificado de operações. Confira Operações de atualização para exibições materializadas.

COPY INTO fornece sintaxe SQL familiar para processamento em lote incremental para arquivos de dados no armazenamento de objetos na nuvem. O comportamento de COPY INTO é semelhante aos padrões aceitos pelo Streaming Tables for Cloud Object Storage, mas nem todas as configurações padrão são equivalentes para todos os formatos de arquivo compatíveis.