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 amplo suporte 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 linhas na interface do usuário do notebook, talvez você não observe todos os dados lidos por consultas de fluxo. Nas cargas de trabalho de produção, você só deve disparar consultas de streaming gravando-as em uma 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
• Cinética
• PubSub
• Pulsar

Você precisa fornecer detalhes de configuração ao inicializar consultas em relação a esses sistemas, que variam dependendo do ambiente configurado e do sistema do qual você escolhe ler. Consulte conectores Padrões no Lakeflow Connect.

Cargas de trabalho comuns que envolvem sistemas de fluxo incluem ingestão de dados no lakehouse e processamento de fluxo para coletar dados em sistemas externos. Para obter mais informações sobre cargas de trabalho de streaming, consulte conceitos de Streaming Estruturado.

Os exemplos a seguir demonstram uma leitura interativa de streaming do 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 em modo de leitura por streaming

O Azure Databricks cria todas as tabelas usando Delta Lake por padrão. Ao executar uma consulta de fluxo em uma tabela Delta, a consulta detecta automaticamente novos registros quando uma versão da tabela é confirmada. Por padrão, as consultas de streaming esperam que as tabelas de origem contenham apenas registros acrescentados. Se você precisar trabalhar com dados de streaming que contenham atualizações e exclusões, o Databricks recomenda usar o Lakeflow Spark Declarative Pipelines e AUTO CDC ... INTO. Confira as APIs AUTO CDC: Simplifique a captura de alterações de dados com pipelines.

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

Python

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

SQL

SELECT * FROM STREAM table_name

Consultar dados no armazenamento de objetos de nuvem com o Carregador Automático

Você pode transmitir dados do armazenamento de objetos de nuvem usando o Carregador Automático, o conector de dados de nuvem do Azure Databricks. Você pode usar o conector com arquivos armazenados em volumes do Catálogo do Unity ou em outros locais de armazenamento de objetos de nuvem. O Databricks recomenda o uso de volumes para gerenciar o acesso aos 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 ingestão de streaming de dados no armazenamento de objetos de nuvem 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 possíveis perdas de dados devido a registros corrompidos ou alterações de esquema.

Para obter mais recomendações sobre como ingerir dados do armazenamento de objetos de nuvem, consulte conectores Standard no Lakeflow Connect.

Os exemplos a seguir demonstram uma leitura interativa de streaming 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')