Configurare un flusso

Un oggetto Stream rappresenta un'origine dati di streaming esterna, ad esempio Apache Kafka. Streams archivia i dettagli della connessione, l'autenticazione, gli schemi e la configurazione di acquisizione. Dopo aver creato un flusso, è possibile farvi riferimento usando le definizioni di Visualizzazione funzionalità per creare funzionalità di streaming in tempo reale.

I flussi hanno nomi in tre parti (catalog.schema.stream_name). L'accesso a uno Stream è regolato dalla relativa tabella di acquisizione associata. Vedi Ingestione e riempimento per i dettagli.

Requisiti

• Per eseguire i comandi del notebook: serverless o un cluster di calcolo classico che esegue Databricks Runtime 17.0 ML o versione superiore.
• Deve essere installato il feature-engineering-client pacchetto Python versione 0.17.0 o superiore.

Connessione alle sorgenti di flusso

Prima di definire le funzionalità di streaming, stabilisci e testa la connessione di una pipeline Lakeflow Streaming al broker Kafka. Feature Store si basa su serverless SDP, il che significa che avrai bisogno di un meccanismo per collegare il tuo classico computo (broker o endpoint) al computo serverless di Databrick. Questo avviene tramite prodotti come privatelink o permettendo che il tuo calcolo classico sia accessibile da internet pubblico.

Creare un flusso

Usare create_stream() per creare un nuovo flusso. Un flusso richiede quattro componenti di configurazione:

• Configurazione sorgente: Specifica la piattaforma di streaming e dettagli specifici della sorgente, come l'abbonamento all'argomento per una sorgente Kafka.
• Configurazione connessione: specifica come connettersi ed eseguire l'autenticazione alla piattaforma di streaming, inclusi server e credenziali bootstrap.
• Configurazione dello schema: definisce la struttura delle chiavi e dei valori dei messaggi.
• Configurazione di acquisizione: specifica dove e come vengono acquisiti i dati del flusso. Vedi Ingestione e riempimento per i dettagli.

Per la configurazione specifica source_config per sorgente e la connessione, insieme a un esempio completo create_stream() , vedi Apache Kafka. Le opzioni di schema e di ingestione sono condivise tra le fonti.

Apache Kafka

Per trasmettere da Apache Kafka, usa KafkaStreamConfig come configurazione sorgente e una connessione Unity Catalog per l'autenticazione. Consulta Streaming su elaborazione serverless e Connettersi ad Apache Kafka per la connettività con Kafka.

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"
        ),
    ),
)

Modalità di abbonamento Kafka

La modalità di sottoscrizione specifica il modo in cui Stream seleziona gli argomenti Kafka da utilizzare. Sono supportate tre modalità:

Autenticazione Kafka

Connessione al catalogo Unity (scelta consigliata)

Usare una connessione del catalogo Unity per eseguire l'autenticazione al cluster Kafka. Questo è l'approccio consigliato per l'autenticazione gestita. Per creare una connessione, vedere Creare una connessione. Il creatore dello Stream deve avere USE CONNECTION sulla connessione. Qualsiasi utente che crea funzionalità usando Stream come origine deve anche disporre di USE CONNECTION per la connessione.

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

La connessione supporta sia l'autenticazione IAM (credenziale di servizio) che l'autenticazione SASL.

IAM (accreditamento di servizio)

Autentica con una credenziale di servizio Unity Catalog, ad esempio per connetterti ad Amazon MSK con IAM. Per creare credenziali del servizio, vedere Creare credenziali del servizio. Imposta il nome della credenziale del servizio con l'opzione credential :

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    credential '<service_credential>'
)

Oltre a USE CONNECTION sulla connessione, le identità che utilizzano la credenziale del servizio necessitano di ACCESS su di essa. Concedi ACCESS la credenziale di servizio referenziato al creatore dello Stream e a qualsiasi identità che materializzi funzionalità con lo Stream. Vedere Concedere le autorizzazioni per usare le credenziali di un servizio per accedere a un servizio cloud esterno.

SASL

L'autenticazione SASL utilizza un nome utente e una password. Impostare sasl_mechanism su uno dei valori seguenti:

• PLAIN
• SCRAM-SHA-256
• SCRAM-SHA-512

Fornisci le credenziali con le opzioni user e password. La connessione memorizza queste credenziali in modo sicuro.

Il seguente esempio utilizza SASL/SCRAM. Per SASL/PLAIN, imposta sasl_mechanism a PLAIN.

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    sasl_mechanism 'SCRAM-SHA-512',
    user '<username>',
    password '<password>'
)

MTLS diretto

Per l'autenticazione mTLS diretta, specificare i file keystore e truststore archiviati in un volume di Unity Catalog, con le password referenziate tramite gli scope dei segreti di Databricks. Per altre informazioni sull'autenticazione SSL con Kafka, vedere Usare SSL per connettersi Azure Databricks a 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"
        ),
    ),
)

Configurazione dello schema

Definire la struttura delle chiavi e dei valori dei messaggi in modo che le definizioni di ingestione e caratteristiche possano leggere i singoli campi. Per le origini Kafka, payload_schema corrisponde al valore del messaggio Kafka (modello value chiave-valore di Kafka) e key_schema corrisponde alla chiave del messaggio Kafka. È necessario specificare almeno uno di payload_schema o key_schema .

Ognuno SchemaConfig accetta uno dei tre formati, che corrispondono a come la sorgente serializza i suoi messaggi: json_schema, avro_schema, o proto_schema. Se non viene fornito alcuno schema per una chiave o un payload, viene considerato come una stringa semplice.

Gli esempi di codice in questa sezione utilizzano schemi dichiarati in linea con DirectSchemas, dove lo schema è fornito come stringa. Per gestire gli schemi utilizzando un registro di schemi esterno, vedi Registro degli schemi per i dettagli.

Schema JSON

Fornire una stringa di schema JSON a json_schema.

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"}'
    ),
)

Schema Avro

Fornisci una stringa dello schema Avro a avro_schema. Sono supportati tipi logici avro, inclusi timestamp-millis, date, e decimal.

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        avro_schema=(
            '{'
            '  "type": "record",'
            '  "name": "Event",'
            '  "fields": ['
            '    {"name": "user_id", "type": "string"},'
            '    {"name": "amount", "type": "double"},'
            '    {"name": "event_time",'
            '     "type": {"type": "long", "logicalType": "timestamp-millis"}}'
            '  ]'
            '}'
        )
    ),
)

Schema Protobuf

Fornire un ProtoSchemaSpec a proto_schema con il testo sorgente Protocol Buffers.proto e il nome del messaggio payload. Importare ProtoSchemaSpec da databricks.feature_engineering.entities.

message_name deve essere il nome del messaggio completamente qualificato, incluso quello package dichiarato nel .proto testo (ad esempio, com.example.Event, non Event). Sono supportate sia la sintassi proto2 che proto3.

google.protobuf.Timestamp e i tipi di involucro scalare (StringValue, Int32Value, e così via) sono supportati, e le loro importazioni vengono risolte automaticamente. Altri tipi ben noti, come Duration, Struct, e Any, sono rifiutati; codifica quei valori come uno scalare o un messaggio supportato invece. I tipi scalari fixed32 e fixed64, nonché map con chiavi non stringa, non sono supportati.

from databricks.feature_engineering.entities import ProtoSchemaSpec

schema_config = DirectSchemas(
    payload_schema=SchemaConfig(
        proto_schema=ProtoSchemaSpec(
            schema_text=(
                'syntax = "proto3";\n'
                'package com.example;\n'
                'import "google/protobuf/timestamp.proto";\n'
                'message Event {\n'
                '  string user_id = 1;\n'
                '  double amount = 2;\n'
                '  google.protobuf.Timestamp event_time = 3;\n'
                '}'
            ),
            message_name="com.example.Event",
        )
    ),
)

Decodifica dei dati tramite schemi

Databricks decodifica ogni messaggio con le funzioni di Spark from_json, from_avro, e from_protobuf . I seguenti comportamenti si applicano sia che dichiari lo schema in linea sia che lo risolvi da un registro di schema:

• Dischi malformati. La decodifica utilizza la modalità PERMISSIVE, quindi un record che non corrisponde al proprio schema viene decodificato come valore nullo invece di mandare in errore il flusso.
• Union Avro. Un'unione di più tipi di record viene decodificata in una struct con un campo per ogni tipo di record, ognuno con il nome del rispettivo record Avro.
• Tipi Protobuf. Gli interi senza segno vengono decodificati come un tipo con segno più ampio (ad esempio, da uint32 a BIGINT e da uint64 a DECIMAL(20,0)), i campi enum vengono decodificati nel rispettivo nome stringa e i tipi wrapper scalari (ad esempio, StringValue e Int32Value) vengono decodificati in una colonna nullable del tipo sottostante.

Registro dello schema

I registri degli schemi memorizzano e versionano gli schemi che i produttori e i consumatori di streaming utilizzano, facendo rispettare le regole di compatibilità man mano che tali schemi si evolvono. Quando viene configurato un registro di schema esterno, Feature Store legge lo schema dal registro e lo utilizza per decodificare il messaggio in streaming. Non dichiari lo schema inline nello Stream quando usi un registro di schema.

Il supporto al registro schema presenta le seguenti limitazioni:

• Supportato solo per gli stream Kafka.
• È supportato solo il Confluent Schema Registry
• Sono supportati solo i formati Avro e Protobuf . Per leggere i messaggi JSON, dichiara invece lo schema inlineato. Vedi schema JSON.
• Ogni Flusso è collegato esattamente a un soggetto Confluente per il valore del messaggio e uno per la chiave del messaggio (se fornita). Gli argomenti di flusso contenenti più record di schema non sono una configurazione supportata. Se il tuo Stream si collega a argomenti che contengono più schemi, i record che non corrispondono allo schema per l'argomento specificato vengono decodificati come nulli.

Connettiti a un registro di schema

Specifica i dettagli di connessione del registro come opzioni nella connessione Kafka Unity Catalog e memorizza la chiave segreta API del registro in un secret scope di Databricks. L'identità run-as dello Stream deve disporre dell'autorizzazione READ sull'ambito del segreto, perché la pipeline di acquisizione legge il segreto in fase di esecuzione. Per come creare e configurare una connessione, vedi Crea una connessione.

Aggiungi le schema_registry_urlopzioni , schema_registry_api_key, e schema_registry_api_secret alla connessione utilizzata per l'autenticazione. Il seguente esempio crea una connessione Kafka che si autentica presso il broker con una credenziale di servizio Unity Catalog e con il registro con una chiave API:

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    credential '<service_credential>',
    schema_registry_url 'https://<registry-host>',
    schema_registry_api_key '<registry_api_key>',
    schema_registry_api_secret secret('<scope>', '<key>')
)

Imposta sia l'opzione schema_registry_api_secret sulla connessione Kafka sia il riferimento segreto sull'ambito dello Stream allo stesso segreto.

Crea un flusso che utilizza un registro di schema

Passa a SchemaRegistryConfig come il schema_config. Fai riferimento al segreto dell'API del registro con api_secret_ref, e identifica l'oggetto e il formato con payload_schema_locator per il valore del messaggio, o key_schema_locator per la chiave del messaggio. Deve essere fornito almeno un localizzatore.

Nota le differenze qui rispetto agli esempi diretti di schema nella sezione Configurazione dello schema . Quando si utilizza un registro degli schemi, non si fornisce lo schema direttamente nello stream a schema_config. Invece, specifichi un SchemaRegistryConfig che identifica lo schema nel registro.

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    KafkaStreamConfig,
    KafkaSubscriptionMode,
    StreamConnectionConfig,
    SchemaRegistryConfig,
    SchemaLocator,
    SchemaLocatorConfluentSchema,
    SchemaLocatorFormat,
    SecretScopeReference,
    IngestionConfig,
    IngestionDestination,
)

client = FeatureEngineeringClient()

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    source_config=KafkaStreamConfig(
        subscription_mode=KafkaSubscriptionMode(subscribe="transactions"),
    ),
    connection_config=StreamConnectionConfig(
        uc_connection_name="my-kafka-connection"
    ),
    schema_config=SchemaRegistryConfig(
        api_secret_ref=SecretScopeReference(
            scope="my_scope", key="sr_api_secret"
        ),
        payload_schema_locator=SchemaLocator(
            confluent_schema=SchemaLocatorConfluentSchema(
                subject="transactions-value"
            ),
            format=SchemaLocatorFormat.FORMAT_AVRO,
        ),
    ),
    ingestion_config=IngestionConfig(
        ingestion_destination=IngestionDestination(
            delta_table_name="my_catalog.my_schema.transactions_ingestion"
        ),
    ),
)

Un soggetto Confluente è l'ambito denominato sotto cui viene registrata la cronologia delle versioni di uno schema e viene applicata la compatibilità. Imposta subject sul nome dell'ambito appropriato, che in genere viene determinato dalla strategia del nome del soggetto:

• TopicNameStrategy (di default, deriva il soggetto dal nome dell'argomento): <topic>-value per il valore e <topic>-key per la chiave. Ad esempio, lo schema di valori per l'argomento transactions utilizza il soggetto transactions-value.
• RecordNameStrategy (deriva l'oggetto dal nome del record dello schema, indipendentemente dall'argomento): il nome del record completamente qualificato, come com.example.Payment. Questo è il namespace e il nome del record per Avro, oppure il package e il nome del messaggio per Protobuf.
• TopicRecordNameStrategy (combina i nomi di argomento e record): <topic>-<fully-qualified-record-name>, come transactions-com.example.Payment.

format è obbligatorio. Impostalo su SchemaLocatorFormat.FORMAT_AVRO o SchemaLocatorFormat.FORMAT_PROTOBUF in modo che corrisponda a come l'argomento viene serializzato.

Evoluzione dello schema

La pipeline di ingestione risolve lo schema corrente del soggetto quando inizia. Quando registri una nuova versione dello schema retrocompatibile sull'oggetto nel registro dello schema, la pipeline in esecuzione continua a usare la versione con cui era iniziata.

Poiché Databricks gestisce la pipeline di ingestione come una pipeline Lakeflow serverless, questa si riavvia periodicamente. Al riavvio successivo, rileva la nuova versione dello schema. Possono essere necessarie fino a una settimana prima che i campi nuovi o modificati vengano visualizzati nella tabella di ingestione.

Per come la pipeline gestisce i record che non corrispondono allo schema attualmente utilizzato, vedi Decodifica dei dati usando gli schemi.

Ingestione e riempimento

Il ingestion_config parametro consente di configurare la modalità di acquisizione e archiviazione dei dati di flusso per il training e la gestione.

L'accesso a un flusso è regolato dalla tabella di inserimento:

• SELECT nella tabella di acquisizione concede l'accesso in lettura allo stream.
• MANAGE nella tabella di inserimento concede l'accesso all'eliminazione.

Per ulteriori informazioni sui privilegi della tabella, vedere Tabella e riferimento ai privilegi di Unity Catalog.

Pipeline di inserimento

Quando viene creato un flusso, Databricks avvia una pipeline di ingestione gestita che legge continuamente i messaggi dal flusso sorgente e li scrive in una tabella Delta (la tabella di ingestione). La pipeline parte dalla posizione più recente nella sorgente e procede continuamente, catturando solo i nuovi messaggi che arrivano dopo la creazione del flusso. Questa tabella di acquisizione viene utilizzata per l'addestramento con funzionalità in streaming. Quando un flusso viene eliminato, vengono eliminate anche la pipeline di inserimento e la tabella di inserimento.

Destinazione di acquisizione

ingestion_destination Specifica il nome della tabella Delta in tre parti in cui vengono scritti i dati del flusso.

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

Schema della tabella di acquisizione

La tabella di ingestione contiene i dati dei messaggi insieme alle colonne dei metadati. Le colonne comuni sono presenti per ogni sorgente; le kafka_* colonne sono presenti solo per un flusso Kafka.

Origine del riempimento retroattivo

Poiché la pipeline forward-fill parte dalla posizione più recente nella sorgente, non cattura i messaggi esistenti prima della creazione del flusso. Per fornire una copertura storica dei dati per l'addestramento, configura una fonte facoltativa di recupero dati pregressi.

Quando viene configurata un'origine di backfill, Databricks esegue un processo una tantum MERGE INTO che copia le righe di backfill nella tabella di ingestione con record_source="backfill". L'operazione MERGE viene eseguita solo dopo che la verifica della sovrapposizione conferma che l'origine del backfill e il flusso di forward-fill hanno timestamp sovrapposti (consulta Sovrapposizione tra il backfill e i dati del flusso live). Se la condizione di sovrapposizione non viene soddisfatta entro 2 giorni, l'operazione MERGE viene eseguita comunque per evitare il blocco illimitato.

La tabella backfill deve includere una stream_record_timestamp colonna di tipo TIMESTAMP nel fuso orario UTC. Le altre colonne di metadati vengono mantenute se presenti nella sorgente di backfill, oppure impostate su NULL in caso contrario. Per Kafka, questi sono kafka_topic, kafka_partition, e kafka_offset.

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"
    ),
)

Sovrapposizione tra il riempimento retroattivo e i dati del flusso in tempo reale

Prima di eseguire un'operazione MERGE tra il backfill e la tabella di acquisizione, un controllo di sovrapposizione confronta i timestamp presenti nelle due tabelle:

• Backfill max: valore massimo stream_record_timestamp nell'origine del riempimento.
• Inserimento min: numero minimo stream_record_timestamp di righe (record_source="stream") nella tabella di inserimento.

Il MERGE procede quando il timestamp più recente del backfill supera di almeno 1 ora il timestamp più vecchio della tabella di ingestione. Questa sovrapposizione garantisce che non vi siano lacune nella tabella di inserimento. Se la condizione di sovrapposizione non viene soddisfatta entro 2 giorni, l'operazione MERGE viene eseguita comunque per evitare il blocco illimitato.

Poiché la pipeline di ingestione parte dalla posizione più recente nella sorgente, cattura i messaggi che arrivano solo dopo la creazione del flusso. L’origine del backfill deve contenere dati che si estendono nell’intervallo temporale di acquisizione, e non solo fino al momento di creazione del flusso.

Ad esempio, se si crea un flusso alle 15:00, la pipeline di forward-fill inizia a leggere i messaggi a partire dalle 15:00. L'origine dei dati di backfill deve includere dati con timestamp che arrivino almeno alle 16:00 (1 ora dopo l'inizio del forward-fill) per superare la verifica di sovrapposizione. Ciò significa che è necessario aggiornare la tabella backfill dopo le 14:00 per assicurarsi che la tabella di inserimento non contenga lacune.

Deduplicazione

Usare deduplication_columns per specificare i percorsi delle colonne per identificare le righe duplicate in fase di acquisizione tra i dati di flusso di backfill e forward-fill. Usare la notazione punto per i campi annidati , ad esempio "value.user_id".

Scegliere le colonne di deduplicazione in base ai dati:

• Se ogni record nel flusso contiene un identificatore univoco ( ad esempio , value.transaction_id), usare tale colonna per la deduplicazione.
• Se l'origine di backfill include le colonne kafka_partition e kafka_offset, usale per identificare in modo univoco ogni record.
• Se non vengono specificate colonne di deduplicazione, la chiave di deduplicazione predefinita è la combinazione completa di key, valuee stream_record_timestamp. Questa operazione non è consigliata perché questa rigorosa corrispondenza dei criteri può causare facilmente duplicati.

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

Gestire i flussi

Ottenere un flusso

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

Elenco dei flussi

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

Impostare include_schemas=True per includere i dettagli completi dello schema. Gli schemi possono essere di grandi dimensioni e ciò potrebbe comportare un'operazione a esecuzione prolungata. Per recuperare gli schemi singolarmente, usare get_streaminvece .

Eliminare un flusso

L'eliminazione di un flusso elimina anche la pipeline di acquisizione e la tabella di acquisizione.

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

Notebook di esempio

Per un esempio completo che crea uno Stream, definisce le funzionalità di streaming e distribuisce su un endpoint di serving, consulta il notebook seguente:

Notebook di avvio rapido sulle visualizzazioni delle funzionalità di streaming

Ottieni il notebook