Consultar dados de streaming

Você pode usar o Azure Databricks para consultar fontes de dados de streaming usando o Streaming Estruturado. O Azure Databricks fornece suporte extensivo para cargas de trabalho de streaming em Python e Scala e dá suporte à maioria das funcionalidades de Streaming Estruturado com SQL.

Os exemplos a seguir demonstram o uso de um coletor de memória para inspeção manual de dados de streaming durante o desenvolvimento interativo em notebooks. Devido aos limites de saída de linha na interface do usuário do bloco de notas, pode não conseguir observar todos os dados lidos por consultas de streaming. Em cargas de trabalho de produção, deve-se acionar consultas em fluxo contínuo apenas gravando-as numa tabela de destino ou sistema externo.

Consultar dados de sistemas de streaming

O Azure Databricks fornece leitores de dados de streaming para os seguintes sistemas de streaming:

• Kafka
• Cinesis
• PubSub
• Pulsar

Você deve fornecer detalhes de configuração ao inicializar consultas nesses sistemas, que variam dependendo do ambiente configurado e do sistema escolhido para ler. Consulte Conectores padrão no Lakeflow Connect.

As cargas de trabalho comuns que envolvem sistemas de streaming incluem a ingestão de dados para o lakehouse e o processamento de fluxo para enviar dados para sistemas externos. Para saber mais sobre cargas de trabalho de streaming, consulte Conceitos de streaming estruturado.

Os exemplos a seguir demonstram uma leitura interativa de streaming a partir de Kafka.

Python

display(spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "latest")
  .load()
)

SQL

SELECT * FROM STREAM read_kafka(
  bootstrapServers => '<server:ip>',
  subscribe => '<topic>',
  startingOffsets => 'latest'
);

Consultar uma tabela através de leitura em streaming

O Azure Databricks cria todas as tabelas usando o Delta Lake por padrão. Quando se realiza uma consulta de streaming numa tabela Delta, a consulta detecta automaticamente novos registos quando uma versão da tabela é confirmada. Por padrão, as consultas de streaming esperam que as tabelas de origem contenham apenas registros anexados. Se você precisar trabalhar com dados de streaming que contenham atualizações e exclusões, o Databricks recomenda o uso de Lakeflow Spark Declarative Pipelines e AUTO CDC ... INTO. Consulte As APIs do AUTO CDC: Simplifique a captura de dados de alteração com pipelines.

Os exemplos a seguir demonstram a execução de uma leitura de streaming interativo de uma tabela:

Python

display(spark.readStream.table("table_name"))

SQL

SELECT * FROM STREAM table_name

Consultar dados no armazenamento de objetos na nuvem com o Auto Loader

Você pode transmitir dados do armazenamento de objetos na nuvem usando o Auto Loader, o conector de dados na nuvem do Azure Databricks. Você pode usar o conector com arquivos armazenados em volumes do Unity Catalog ou outros locais de armazenamento de objetos na nuvem. O Databricks recomenda o uso de volumes para gerenciar o acesso a dados no armazenamento de objetos na nuvem. Consulte Conectar-se a fontes de dados e serviços externos.

O Azure Databricks otimiza esse conector para a ingestão de streaming de dados no armazenamento de objetos na nuvem que são armazenados em formatos estruturados, semiestruturados e não estruturados populares. O Databricks recomenda armazenar dados ingeridos em um formato quase bruto para maximizar a taxa de transferência e minimizar a perda potencial de dados devido a registros corrompidos ou alterações de esquema.

Para obter mais recomendações sobre como ingerir dados do armazenamento de objetos na nuvem, consulte Conectores padrão no Lakeflow Connect.

Os exemplos a seguir demonstram uma leitura de streaming interativo de um diretório de arquivos JSON em um volume:

Python

display(spark.readStream.format("cloudFiles").option("cloudFiles.format", "json").load("/Volumes/catalog/schema/volumes/path/to/files"))

SQL

SELECT * FROM STREAM read_files('/Volumes/catalog/schema/volumes/path/to/files', format => 'json')