Configurar uma transmissão

Um Stream representa uma fonte externa de dados de streaming, como o Apache Kafka. Os fluxos armazenam detalhes de ligação, autenticação, esquemas e configuração de ingestão. Depois de um fluxo ser criado, pode referenciá-lo utilizando definições de Feature View para criar funcionalidades de streaming em tempo real.

Os cursos de água têm nomes em três partes (catalog.schema.stream_name). O acesso a um Stream é regulado pela sua tabela de ingestão associada. Consulte Ingestão e retropreenchimento para mais detalhes.

Requisitos

• Para executar comandos de bloco de notas: sem servidor ou um cluster de computação clássico com o Databricks Runtime 17.0 ML ou superior.
• O feature-engineering-client pacote Python versão 0.16.0 ou superior deve ser instalado.

Criar um fluxo

Use create_stream() para criar um novo Stream. Um Fluxo requer quatro componentes de configuração:

• Configuração da fonte: Especifica a plataforma de streaming (por exemplo, Kafka) e detalhes específicos da fonte (como subscrição de tópicos para Kafka).
• Configuração de ligação: Especifica como se ligar e autenticar à plataforma de streaming, incluindo servidores bootstrap e credenciais.
• Configuração do esquema: Define a estrutura das chaves e valores das mensagens.
• Configuração de ingestão: Especifica onde e como os dados do fluxo são ingeridos. Consulte Ingestão e retropreenchimento para mais detalhes.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    KafkaStreamConfig,
    KafkaSubscriptionMode,
    StreamConnectionConfig,
    DirectSchemas,
    SchemaConfig,
    IngestionConfig,
    IngestionDestination,
    StreamBackfillSource,
)

client = FeatureEngineeringClient()

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    source_config=KafkaStreamConfig(
        subscription_mode=KafkaSubscriptionMode(subscribe="events-topic"),
    ),
    connection_config=StreamConnectionConfig(
        uc_connection_name="my-kafka-connection"
    ),
    schema_config=DirectSchemas(
        payload_schema=SchemaConfig(
            json_schema=(
                '{'
                '  "type": "object",'
                '  "properties": {'
                '    "transaction_id": {"type": "string"},'
                '    "user_id": {"type": "string"},'
                '    "amount": {"type": "number"},'
                '    "event_time": {"type": "string", "format": "date-time"}'
                '  }'
                '}'
            )
        ),
    ),
    ingestion_config=IngestionConfig(
        ingestion_destination=IngestionDestination(
            delta_table_name="my_catalog.my_schema.events_ingestion"
        ),
    ),
)

Ligação às fontes de ribeiro

Antes de definir funcionalidades de streaming, estabeleça e teste uma ligação de pipeline Lakeflow de streaming ao seu broker Kafka. Consulte Streaming na computação sem servidor e Ligar ao Apache Kafka.

Para streaming gerido pela AWS (Amazon MSK), veja Conectividade privada sem servidor para a Amazon MSK. Para detalhes sobre as opções de autenticação Kafka, consulte Autenticação.

Authentication

Ligação ao Catálogo Unity (recomendado)

Use uma ligação ao Unity Catalog para autenticar no seu cluster Kafka. Esta é a abordagem recomendada para autenticação gerida. Para criar uma ligação, veja Criar uma ligação.

connection_config = StreamConnectionConfig(
    uc_connection_name="my-kafka-connection"
)

MTLS direto

Para autenticação mTLS direta, forneça ficheiros keystore e truststore armazenados num volume do Unity Catalog, com palavras-passe referenciadas através dos escopos secretos do Databrick. Para mais informações sobre autenticação SSL com Kafka, veja Usar SSL para ligar o Azure Databricks ao Kafka.

from databricks.feature_engineering.entities import (
    DirectMtlsConfig,
    MtlsConfig,
    SecretScopeReference,
)

connection_config = DirectMtlsConfig(
    bootstrap_servers="broker1:9092,broker2:9092",
    mtls_config=MtlsConfig(
        keystore_location="/Volumes/my_catalog/my_schema/my_volume/keystore.jks",
        keystore_password_ref=SecretScopeReference(
            scope="my_scope", key="keystore_password"
        ),
        key_password_ref=SecretScopeReference(
            scope="my_scope", key="key_password"
        ),
        truststore_location="/Volumes/my_catalog/my_schema/my_volume/truststore.jks",
        truststore_password_ref=SecretScopeReference(
            scope="my_scope", key="truststore_password"
        ),
    ),
)

SASL

A autenticação SASL (tanto SASL/SCRAM como SASL/PLAIN) não é suportada durante a pré-visualização.

Modos de subscrição

O modo de subscrição especifica como o Stream seleciona os tópicos Kafka para consumir. São suportados três modos:

Configuração do esquema

Defina a estrutura das chaves e valores das mensagens usando o formato JSON Schema . Para fontes de Kafka, payload_schema corresponde ao valor da mensagem de Kafka (o value no modelo chave-valor de Kafka) e key_schema corresponde à chave de mensagem de Kafka. Pelo menos um dos payload_schema ou key_schema deve ser fornecido.

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        json_schema=(
            '{'
            '  "type": "object",'
            '  "properties": {'
            '    "user_id": {"type": "string"},'
            '    "amount": {"type": "number"},'
            '    "event_time": {"type": "string"}'
            '  }'
            '}'
        )
    ),
    key_schema=SchemaConfig(
        json_schema='{"type": "string"}'
    ),
)

Se não for fornecido um esquema para uma chave ou carga útil, é tratado como uma cadeia simples.

Ingestão e preenchimento retroativo

O ingestion_config parâmetro configura como os dados do fluxo são capturados e armazenados para treino e serviço.

O acesso a um Stream é regido pela tabela de ingestão:

• SELECT na tabela de ingestão concede acesso de leitura ao Stream.
• MANAGE na tabela de ingestão concede permissões de eliminação.

Para mais informações sobre privilégios da tabela, consulte Tabela e referência de privilégios do Unity Catalog.

Canal de ingestão

Quando um fluxo é criado, o Databricks inicia um pipeline de ingestão gerido que lê continuamente mensagens do tópico Kafka e as escreve numa tabela Delta (a tabela de ingestão). A canalização começa no offset mais recente do Kafka e é executada continuamente, captando apenas as novas mensagens que chegam depois de o fluxo ser criado. Esta tabela de ingestão é usada para treino com funcionalidades de streaming. Quando um fluxo é eliminado, o seu canal de ingestão e a tabela de ingestão também são eliminados.

Destino de ingestão

O ingestion_destination especifica o nome da tabela Delta de três partes onde os dados de fluxo são gravados.

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
)

Esquema da tabela de ingestão

A tabela de ingestão contém os dados da mensagem juntamente com colunas de metadados:

Fonte de preenchimento

Como o pipeline de preenchimento progressivo começa a partir do offset do Kafka mais recente, ele não capta mensagens que já existiam antes de a stream ser criada. Para fornecer cobertura histórica de dados para treino, configure uma fonte opcional de preenchimento.

Quando é configurada uma origem de backfill, o Databricks executa uma tarefa única MERGE INTO que copia as linhas de backfill para a tabela de ingestão com record_source="backfill". O MERGE só funciona depois de o verificador de sobreposição confirmar que a fonte de preenchimento e o fluxo de preenchimento direto têm carimbos temporais sobrepostos (ver Sobreposição entre dados de preenchimento e transmissão em direto). Se a condição de sobreposição não for satisfeita no prazo de 2 dias, a MERGE é executada ainda assim para evitar bloqueios indefinidos.

A tabela de preenchimento deve incluir uma stream_record_timestamp coluna do tipo TIMESTAMP no fuso horário UTC. Outras colunas de metadados de Kafka (kafka_topic, kafka_partition, kafka_offset) são passadas se estiverem presentes na fonte de preenchimento, ou definidas como NULL de outra forma.

from databricks.feature_engineering.entities import StreamBackfillSource

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
    backfill_source=StreamBackfillSource(
        delta_table_name="my_catalog.my_schema.historical_events"
    ),
)

Sobreposição entre os dados de preenchimento e transmissão ao vivo

Antes de executar uma operação MERGE entre o backfill e a tabela de ingestão, uma verificação de sobreposição compara as marcas temporais das duas tabelas:

• Máximo de preenchimento: O máximo stream_record_timestamp na fonte de preenchimento.
• Mínimo de ingestão: O mínimo stream_record_timestamp de linhas (record_source="stream") na tabela de ingestão.

A operação MERGE é executada quando a marca temporal mais recente do preenchimento retroativo é superior à marca temporal mais antiga da tabela de ingestão em, pelo menos, 1 hora. Esta sobreposição garante que não existem lacunas na tabela de ingestão. Se a condição de sobreposição não for satisfeita no prazo de 2 dias, a MERGE é executada ainda assim para evitar bloqueios indefinidos.

Como o pipeline de ingestão começa a partir do último offset do Kafka, apenas captura mensagens que chegam depois de o fluxo ser criado. A sua origem de preenchimento retroativo deve conter dados que abranjam o intervalo de tempo de ingestão — e não apenas até ao momento de criação do fluxo.

Por exemplo, se criar um fluxo de dados às 15:00, a canalização de preenchimento progressivo começa a ler mensagens a partir das 15:00. A sua origem do preenchimento retroativo deve incluir dados com marcas temporais até, pelo menos, às 16:00 (1 hora após o início do preenchimento progressivo) para cumprir a verificação de sobreposição. Isto significa que deve atualizar a sua tabela de preenchimento após as 16:00 para garantir que a tabela de ingestão não tem lacunas.

Deduplication

Utilize deduplication_columns para especificar os caminhos das colunas para identificar linhas duplicadas durante a ingestão entre dados de fluxo de backfill e de forward-fill. Use notação de pontos para campos aninhados (por exemplo, "value.user_id").

Escolha colunas de deduplicação com base nos seus dados:

• Se cada registo no seu fluxo contiver um identificador único (por exemplo, value.transaction_id), use essa coluna para a deduplicação.
• Se a sua origem de preenchimento retroativo incluir as colunas kafka_partition e kafka_offset, utilize-as para identificar cada registo de forma única.
• Se não forem especificadas colunas de deduplicação, a chave de deduplicação por defeito é a combinação completa de key, value, e stream_record_timestamp. Isto não é recomendado, pois esta correspondência rigorosa de critérios pode facilmente levar a duplicados.

ingestion_config = IngestionConfig(
    ingestion_destination=IngestionDestination(
        delta_table_name="my_catalog.my_schema.events_ingestion"
    ),
    deduplication_columns=["value.transaction_id"],
)

Gerir transmissões

Obter uma transmissão

stream = client.get_stream(name="my_catalog.my_schema.my_stream")

Listar fluxos

streams = client.list_streams(
    catalog_name="my_catalog",
    schema_name="my_schema",
    max_results=50,
    include_schemas=False,
)

Defina include_schemas=True para incluir os detalhes completos do esquema. Os esquemas podem ser grandes e isso pode resultar numa operação de longa duração. Para recuperar esquemas individualmente, use get_stream.

Excluir uma transmissão

Eliminar um fluxo também elimina o seu pipeline de ingestão e a tabela de ingestão.

client.delete_stream(name="my_catalog.my_schema.my_stream")

Bloco de notas de exemplo

Para um exemplo de ponta a ponta que cria um Stream, define funcionalidades de streaming e é implementado num endpoint de serviço, veja o seguinte caderno:

Notebook de introdução rápida do Streaming Feature Views

Obter caderno