Configurar um fluxo

Um Stream representa uma fonte de dados de streaming externa, como o Apache Kafka. Os fluxos armazenam detalhes de conexão, autenticação, esquemas e configuração de ingestão. Depois que um stream é criado, você pode fazer referência a ele em definições de Feature View para criar features de streaming em tempo real.

Os fluxos têm nomes de três partes (catalog.schema.stream_name). O acesso a um Stream é regido por sua tabela de ingestão associada. Consulte Ingestão e preenchimento retroativo para obter detalhes.

Requirements

• Para executar comandos de notebook: sem servidor ou um cluster de computação clássico executando 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 Stream requer quatro componentes de configuração:

• Configuração de origem: especifica a plataforma de streaming (por exemplo, Kafka) e detalhes específicos da origem (como assinatura de tópico para Kafka).
• Configuração de conexão: especifica como se conectar e autenticar à plataforma de streaming, incluindo servidores de inicialização e credenciais.
• Configuração de esquema: define a estrutura de chaves e valores de mensagem.
• Configuração de ingestão: especifica onde e como os dados de fluxo são ingeridos. Consulte Ingestão e preenchimento retroativo para obter 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"
        ),
    ),
)

Conectando-se a fontes de streaming

Antes de definir funcionalidades de streaming, conecte e teste uma conexão de pipeline de streaming do Lakeflow com seu broker do Kafka. Consulte Streaming na computação sem servidor e Conexão com o Apache Kafka.

Para o AWS managed streaming (Amazon MSK), consulte conectividade privada sem servidor com o Amazon MSK. Para obter detalhes sobre as opções de autenticação do Kafka, consulte Autenticação.

Autenticação

Conexão com o Unity Catalog (recomendado)

Use uma conexão do Unity Catalog para autenticar no cluster do Kafka. Essa é a abordagem recomendada para autenticação gerenciada. Para criar uma conexão, consulte Criar uma conexão.

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

MTLS direto

Para autenticação direta do mTLS, forneça arquivos de repositório de chaves e de armazenamento confiável armazenados em um volume do Unity Catalog, com senhas referenciadas por meio de escopos secretos do Databricks. Para obter mais informações sobre a autenticação SSL com o Kafka, consulte Usar o SSL para se conectar 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 (SASL/SCRAM e SASL/PLAIN) não é compatível na versão preliminar.

Modos de assinatura

O modo de assinatura especifica como o Stream seleciona os tópicos do Kafka dos quais consumirá. Há suporte para três modos:

Configuração de esquema

Defina a estrutura de chaves de mensagem e valores usando o formato de esquema JSON . Para fontes Kafka, payload_schema corresponde ao valor da mensagem do Kafka (o value no modelo de chave-valor do Kafka) e key_schema corresponde à chave da mensagem do Kafka. Pelo menos um entre payload_schema e 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 nenhum esquema for fornecido para uma chave ou conteúdo, ele será tratado como uma cadeia de caracteres simples.

Ingestão e preenchimento retroativo

O ingestion_config parâmetro configura como os dados de fluxo são capturados e armazenados para treinamento 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 acesso de exclusão.

Para mais informações sobre privilégios de 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 gerenciada que lê continuamente as mensagens do tópico do Kafka e as grava em uma tabela Delta (a tabela de ingestão). O pipeline começa a partir do offset mais recente do Kafka e é executado continuamente, capturando apenas as novas mensagens que chegam após a criação do fluxo. Essa tabela de ingestão é usada para treinamento com recursos de streaming. Quando um fluxo é excluído, seu pipeline de ingestão e sua tabela de ingestão também são excluídos.

Destino de ingestão

O ingestion_destination especifica o nome da tabela Delta em três partes na qual os dados de streaming são gravados.

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

Esquema de tabela de ingestão

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

Origem do preenchimento retroativo

Como o pipeline de preenchimento progressivo começa a partir do offset mais recente do Kafka, ele não captura mensagens anteriores à criação do stream. Para fornecer cobertura de dados históricos para treinamento, configure uma fonte de backfill opcional.

Quando uma fonte de provisionamento é configurada, o Databricks executa um trabalho único MERGE INTO que copia linhas de provisionamento para a tabela de ingestão com record_source="backfill". O MERGE é executado somente depois que o verificador de sobreposição confirma que a origem do provisionamento e o fluxo de preenchimento avançado têm carimbos de data/hora sobrepostos (consulte Sobreposição entre os dados de provisionamento e de transmissão ao vivo). Se a condição de sobreposição não for atendida em até 2 dias, o MERGE é executado mesmo assim para evitar bloqueio indefinido.

A tabela de provisionamento deve incluir uma coluna stream_record_timestamp do tipo TIMESTAMP com fuso horário UTC. Outras colunas de metadados do Kafka (kafka_topic, kafka_partition, kafka_offset) são repassadas se estiverem presentes na origem de backfill; caso contrário, são definidas como NULL.

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 preenchimento retroativo e dados em fluxo ao vivo

Antes de executar um MERGE entre o provisionamento e a tabela de ingestão, uma verificação de sobreposição compara os timestamps das duas tabelas:

• Provisionamento máximo: o máximo stream_record_timestamp na origem do provisionamento.
• Ingestão mínima: o número mínimo stream_record_timestamp de linhas (record_source="stream") na tabela de ingestão.

A operação MERGE é executada quando o timestamp mais recente do provisionamento excede o timestamp mais antigo da tabela de ingestão em pelo menos 1 hora. Essa sobreposição garante que não haja lacunas na tabela de ingestão. Se a condição de sobreposição não for atendida em até 2 dias, o MERGE é executado mesmo assim para evitar bloqueio indefinido.

Como o pipeline de ingestão começa a partir do offset mais recente do Kafka, ele captura apenas as mensagens que chegam após a criação do fluxo. Sua fonte de provisionamento deve conter dados que abranjam o intervalo de tempo de ingestão, não apenas até o momento de criação do fluxo.

Por exemplo, se você criar um stream às 15h, o pipeline de preenchimento progressivo começará a ler mensagens a partir das 15h. Sua fonte de provisionamento deve incluir dados com registros de data e hora até pelo menos 16h (1 hora após o início do preenchimento progressivo) para atender à verificação de sobreposição. Isso significa que você deve atualizar sua tabela de backfill após as 16h para garantir que a tabela de ingestão não tenha lacunas.

Eliminação de duplicação

Use deduplication_columns para especificar os caminhos das colunas para identificar linhas duplicadas durante a ingestão entre dados de provisionamento e dados de fluxo de preenchimento progressivo. Usar notação de ponto para campos aninhados (por exemplo, "value.user_id").

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

• Se cada registro em seu fluxo contiver um identificador exclusivo (por exemplo, value.transaction_id), use essa coluna para eliminação de duplicação.
• Se a fonte de provisionamento incluir as colunas kafka_partition e kafka_offset, use-as para identificar cada registro de forma exclusiva.
• Se nenhuma coluna de eliminação de duplicação for especificada, a chave de eliminação de duplicação padrão será a combinação completa de key, valuee stream_record_timestamp. Isso não é recomendado, pois essa correspondência de critérios estritos pode facilmente levar a duplicatas.

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

Gerenciar fluxos

Obter um fluxo

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 detalhes completos do esquema. Esquemas podem ser grandes, o que pode resultar em uma operação demorada. Para recuperar esquemas individualmente, use get_stream.

Excluir um fluxo

A exclusão de uma transmissão também exclui o pipeline de ingestão e a tabela de ingestão.

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

Notebook de exemplo

Para ver um exemplo completo que cria um Stream, define recursos de streaming e faz a implantação em um endpoint de serviço, consulte o seguinte notebook:

Bloco de anotações de início rápido exibições de recursos de transmissão

Obter laptop