Een Stream instellen

Belangrijk

Deze functie bevindt zich in openbare preview-versie. Werkruimtebeheerders kunnen de toegang tot deze functie beheren vanaf de pagina Previews . Zie Azure Databricks previews beheren.

Een Stream vertegenwoordigt een externe streaminggegevensbron, zoals Apache Kafka. Streams slaan verbindingsdetails, verificatie, schema's en opnameconfiguratie op. Nadat een stream is gemaakt, kunt u ernaar verwijzen met behulp van functieweergavedefinities om realtime streamingfuncties te maken.

Streams hebben driedelige namen (catalog.schema.stream_name). Toegang tot een Stream wordt bepaald door de bijbehorende opnametabel. Zie Opname en backfill voor meer informatie.

Requirements

Voor het uitvoeren van notebookopdrachten: serverloos of een klassiek rekencluster met Databricks Runtime 17.0 ML of hoger.
De feature-engineering-client Python-pakketversie 0.16.0 of hoger moet zijn geïnstalleerd.

Een stream maken

Gebruik create_stream() om een nieuwe Stream te maken. Voor een Stream zijn vier configuratieonderdelen vereist:

Bronconfiguratie: Hiermee geeft u het streamingplatform (bijvoorbeeld Kafka) en bronspecifieke details (zoals onderwerpabonnement voor Kafka) op.
Verbindingsconfiguratie: Hiermee geeft u op hoe u verbinding maakt en verifieert met het streamingplatform, inclusief bootstrapservers en referenties.
Schemaconfiguratie: Definieert de structuur van berichtsleutels en -waarden.
Opnameconfiguratie: Hiermee geeft u op waar en hoe streamgegevens worden opgenomen. Zie Opname en backfill voor meer informatie.

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

Verbinding maken met streambronnen

Voordat u streamingfuncties definieert, verbindt en test u een streaming Lakeflow-pijplijnverbinding met uw Kafka-broker. Zie Streaming op serverloze compute en Verbinding maken met Apache Kafka.

Zie Serverloze privéconnectiviteit met Amazon MSK voor beheerde AWS-streaming (Amazon MSK). Zie Verificatie voor meer informatie over kafka-verificatieopties.

Verificatie

Unity Catalog-verbinding (aanbevolen)

Gebruik een Unity Catalog-verbinding om te verifiëren bij uw Kafka-cluster. Dit is de aanbevolen methode voor beheerde verificatie. Zie Een verbinding maken om een verbinding te maken.

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

Direct mTLS

Geef voor directe mTLS-authenticatie keystore- en truststorebestanden op die zijn opgeslagen in een Unity Catalog-volume, waarvan de wachtwoorden worden opgegeven via Databricks secret scopes. Zie SSL gebruiken om Azure Databricks met Kafka te verbinden voor meer informatie over SSL-authenticatie met 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

SASL-verificatie (zowel SASL/SCRAM als SASL/PLAIN) wordt niet ondersteund tijdens de preview.

Abonnementsmodi

De abonnementsmodus geeft aan hoe de stream Kafka-topics selecteert om van te consumeren. Er worden drie modi ondersteund:

Mode	Description	Voorbeeld
`subscribe`	Door komma's gescheiden lijst met onderwerpnamen	`KafkaSubscriptionMode(subscribe="topic1,topic2")`
`subscribe_pattern`	Java regex-patroon dat overeenkomt met onderwerpnamen	`KafkaSubscriptionMode(subscribe_pattern="events-.*")`
`assign`	JSON die onderwerppartitietoewijzingen specificeert	`KafkaSubscriptionMode(assign='{"my-topic": [0, 1, 2]}')`

Schemaconfiguratie

Definieer de structuur van berichtsleutels en -waarden met behulp van de JSON-schema-indeling . Voor Kafka-bronnen payload_schema komt dit overeen met de Kafka-berichtwaarde (het value sleutelwaardemodel van Kafka) en key_schema komt deze overeen met de Kafka-berichtsleutel. Ten minste één van payload_schema of key_schema moet worden opgegeven.

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

Als er geen schema is opgegeven voor een sleutel of payload, worden deze behandeld als een eenvoudige tekenreeks.

Inname en terugvulling

De ingestion_config parameter configureert hoe streamgegevens worden vastgelegd en opgeslagen voor training en bediening.

Toegang tot een Stream wordt beheerd door de opnametabel:

SELECT op de opnametabel verleent leestoegang tot de Stream.
MANAGE in de opnametabel verleent verwijderingstoegang.

Zie de naslaginformatie over de bevoegdheden van Table en Unity Catalog voor meer informatie over tabelbevoegdheden.

Opnamepijplijn

Wanneer een stream wordt gemaakt, start Databricks een beheerde opnamepijplijn die continu berichten uit het Kafka-onderwerp leest en schrijft naar een Delta-tabel (de opnametabel). De pijplijn begint vanaf de meest recente Kafka-offset en wordt continu uitgevoerd, waarbij alleen nieuwe berichten worden vastgelegd die binnenkomen nadat de stream is gemaakt. Deze innametabel wordt gebruikt voor het trainen met streamingfuncties. Wanneer een stream wordt verwijderd, worden de opnamepijplijn en opnametabel ook verwijderd.

Opnamebestemming

De ingestion_destination specificeert de driedelige naam van de Delta-tabel waarnaar streamgegevens worden weggeschreven.

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

Opnametabelschema

De innametabel bevat de berichtdata samen met kolommen met metagegevens:

Column	Typ	Description
`key`	Varieert (van `key_schema`)	De Kafka-berichtsleutel, gestructureerd volgens het schema dat u hebt opgegeven.
`value`	Varieert (van `payload_schema`)	De Kafka-berichtwaarde (payload), gestructureerd volgens het schema dat u hebt opgegeven.
`stream_record_timestamp`	`TIMESTAMP`	De tijdstempel van het record. Voor forward-fillgegevens is dit de opnametijdstempel van de Kafka-broker. Voor backfillgegevens wordt dit door de klant verstrekt.
`kafka_topic`	`STRING`	De Kafka-topic waaruit het record is gelezen.
`kafka_partition`	`INT`	De Kafka-partitie waaruit het gegevensrecord is gelezen.
`kafka_offset`	`LONG`	De Kafka-offset van het gegevensrecord binnen zijn partitie.
`record_source`	`STRING`	Ofwel `"stream"` (voorwaarts aangevuld vanuit de live-Kafka-stream) ofwel `"backfill"` (vanuit de backfillbron).

Bron voor terugvullen

Omdat de forward-fill-pijplijn begint bij de meest recente Kafka-offset, pikt deze geen berichten op die al bestonden voordat de stream werd aangemaakt. Als u historische gegevensdekking wilt bieden voor training, configureert u een optionele backfillbron.

Wanneer een backfill-bron is geconfigureerd, voert Databricks een eenmalige MERGE INTO taak uit die backfill-rijen naar de inleestabel kopieert met record_source="backfill". De MERGE wordt pas uitgevoerd nadat de overlappingscontrole heeft bevestigd dat de backfillbron en de stroom voor doorsturen overlappende tijdstempels hebben (zie Overlapping tussen backfill- en livestreamgegevens). Als niet binnen 2 dagen aan de overlapvoorwaarde wordt voldaan, gaat de MERGE toch door om te voorkomen dat deze voor onbepaalde tijd wordt geblokkeerd.

De achtervultabel moet een stream_record_timestamp kolom van het type TIMESTAMP bevatten in de UTC-tijdzone. Andere Kafka-metagegevenskolommen (kafka_topic, kafka_partition, kafka_offset) worden doorgegeven als deze aanwezig zijn in de backfillbron, of anders ingesteld op 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"
    ),
)

Overlapping tussen backfill- en livestreamgegevens

Voordat u een MERGE uitvoert tussen backfill en de innametabel, vergelijkt een overlappingscontrole de tijdstempels tussen de twee tabellen:

Backfill max: het maximum stream_record_timestamp in de backfill-bron.
Inname min: het minimumaantal rijen (stream_record_timestamp) in de innametabel.

De MERGE wordt uitgevoerd wanneer de laatste tijdstempel van de backfill minstens 1 uur later valt dan de vroegste tijdstempel van de innametabel. Deze overlapping zorgt ervoor dat er geen hiaten in de opnametabel zijn. Als niet binnen 2 dagen aan de overlapvoorwaarde wordt voldaan, gaat de MERGE toch door om te voorkomen dat deze voor onbepaalde tijd wordt geblokkeerd.

Omdat de opnamepijplijn begint vanaf de meest recente Kafka-offset, worden alleen berichten vastgelegd die binnenkomen nadat de stream is gemaakt. Uw backfill-bron moet gegevens bevatten die doorlopen tot in het innametijdsbereik, niet slechts tot het moment waarop de stream is gemaakt.

Als u bijvoorbeeld om 13:00 uur een stream maakt, begint de pijplijn voor doorsturen met het lezen van berichten vanaf 15:00 uur. Uw bron voor backfill moet gegevens bevatten met tijdstempels die doorlopen tot ten minste 16:00 uur (1 uur na de start van forward-fill) om aan de overlapcontrole te voldoen. Dit betekent dat u uw backfill-tabel na 16:00 uur moet bijwerken om ervoor te zorgen dat de opnametabel geen hiaten bevat.

Ontdubbeling

Gebruik deduplication_columns om kolompaden op te geven voor het identificeren van dubbele rijen tijdens de inname tussen backfill- en forward-fill-streamgegevens. Gebruik punt notatie voor geneste velden (bijvoorbeeld "value.user_id").

Kies ontdubbelingskolommen op basis van uw gegevens:

Als elke record in uw stream een unieke id bevat (bijvoorbeeld value.transaction_id), gebruikt u die kolom voor ontdubbeling.
Als uw backfill-bron kafka_partition en kafka_offset-kolommen bevat, gebruikt u die om elk record uniek te identificeren.
Als er geen ontdubbelingskolommen zijn opgegeven, is de standaardontdubbelingssleutel de volledige combinatie van key, valueen stream_record_timestamp. Dit wordt niet aanbevolen omdat deze strikte criteriakoppeling eenvoudig kan leiden tot duplicaten.

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

Streams beheren

Een stream ophalen

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

Streams weergeven

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

Stel include_schemas=True in zodat alle schemadetails worden opgenomen. Schema's kunnen groot zijn en dit kan leiden tot een langdurige bewerking. Als u in plaats daarvan schema's afzonderlijk wilt ophalen, gebruikt u get_stream.

Een stream verwijderen

Als u een stream verwijdert, worden ook de innamepijplijn en innametabel verwijderd.

Warning

Modellen of functies die naar de verwijderde stream verwijzen, hebben geen toegang meer tot de onderliggende stroomgegevens. Maak een kopie van de opnametabel voordat u deze gegevens verwijdert als u deze gegevens nodig hebt, maar de stream niet meer nodig hebt.

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

Voorbeeld van notebook

Zie het volgende notebook voor een volledig voorbeeld waarin een Stream wordt gemaakt, streamingfuncties worden gedefinieerd en wordt geïmplementeerd naar een serving-eindpunt:

Notebook voor snelstart van streamingfeatureweergaven

Notebook krijgen

Feedback

Is deze pagina nuttig?

Last updated on 2026-07-13