Een Stream instellen

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 Inname en backfill voor details.

Requirements

• Voor het uitvoeren van notebookopdrachten: serverloos of een klassiek rekencluster met Databricks Runtime 17.0 ML of hoger.
• Het feature-engineering-client Python-pakket versie 0.18.0 of hoger moet geïnstalleerd zijn.

Verbinding maken met streambronnen

Voordat u streamingfuncties definieert, verbindt en test u een streaming Lakeflow-pijplijnverbinding met uw Kafka-broker. Feature Store is afhankelijk van serverless SDP, wat betekent dat je een mechanisme nodig hebt om je klassieke compute (broker of endpoint) te verbinden met Databricks serverless compute. Dit gebeurt via producten zoals privatelink of door je klassieke compute toegankelijk te maken vanaf het publieke internet.

Een stream maken

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

• Bronconfiguratie: Specificeert het streamingplatform en bronspecifieke details, zoals het onderwerpabonnement voor een Kafka-bron.
• 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 Inname en backfill voor details.

Voor de bron-specifieke source_config en verbindingsopstelling, samen met een volledig create_stream() voorbeeld, zie Apache Kafka. De schema - en invoeropties worden gedeeld tussen bronnen.

Apache Kafka

Om te streamen vanaf Apache Kafka, gebruik KafkaStreamConfig als bronconfiguratie en een Unity Catalog-verbinding voor authenticatie. Zie Streaming op serverloze rekenkracht en Verbinding maken met Apache Kafka voor connectiviteit met 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"
        ),
    ),
)

Kafka-abonnementsmodi

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

Kafka-authenticatie

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. De maker van de Stream moet beschikken over USE CONNECTION voor de verbinding. Elke gebruiker die features materialiseert waarbij de Stream als bron wordt gebruikt, moet ook USE CONNECTION op de verbinding hebben.

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

De verbinding ondersteunt zowel IAM (service-credential) als SASL-authenticatie.

IAM (serviceaanmeldingsgegeven)

Authenticatie met een Unity Catalog servicecredential, bijvoorbeeld om verbinding te maken met Amazon MSK via IAM. Zie Servicereferenties maken om een servicereferentie te creëren. Stel de naam van de service-credential in met de credential optie:

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

Naast USE CONNECTION op de verbinding, hebben identiteiten die het servicecredential gebruiken ook ACCESS daarop nodig. Verleen ACCESS het genoemde service-credential aan de maker van de stream en aan elke identiteit die functies met de stream materialiseert. Zie Machtigingen verlenen om een servicereferentie te gebruiken voor toegang tot een externe cloudservice.

SASL

SASL-authenticatie gebruikt een gebruikersnaam en wachtwoord. Stel sasl_mechanism in op een van de volgende:

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

Geef de aanmeldingsgegevens op via de opties user en password. De verbinding slaat deze gegevens veilig op.

Het volgende voorbeeld gebruikt SASL/SCRAM. Voor SASL/PLAIN, zet sasl_mechanism op 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>'
)

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

Schemaconfiguratie

Definieer de structuur van berichtsleutels en -waarden zodat ingestie- en feature-definities individuele velden kunnen lezen. 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.

Elke SchemaConfig vorm accepteert een van drie formaten, die overeenkomen met hoe de bron zijn berichten serialiseert: json_schema, avro_schema, of proto_schema. Als er geen schema is opgegeven voor een sleutel of payload, worden deze behandeld als een eenvoudige tekenreeks.

De codevoorbeelden in deze sectie gebruiken schema's die inline zijn gedeclareerd met DirectSchemas, waarbij het schema als string wordt geleverd. Om schema's te beheren met een extern schemaregister, zie Schemaregister voor details.

JSON-schema

Geef een JSON-schemastring aan 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"}'
    ),
)

Avro-schema

Geef een tekenreeks voor het Avro-schema op voor avro_schema. Avro-logische types worden ondersteund, waaronder timestamp-millis, date, en 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"}}'
            '  ]'
            '}'
        )
    ),
)

Protobuf-schema

Geef een ProtoSchemaSpec aan proto_schema samen met de Protocol Buffers.proto-brontekst en de naam van het payload-bericht. Importeren ProtoSchemaSpec uit databricks.feature_engineering.entities.

message_name moet de volledig gekwalificeerde berichtnaam zijn, inclusief de package declaratie in de .proto tekst (bijvoorbeeld com.example.Event, niet Event). Zowel proto2- als proto3-syntaxis worden ondersteund.

google.protobuf.Timestamp en de scalair wrappertypes (StringValue, Int32Value, enzovoort) worden ondersteund, en hun importen worden automatisch opgelost. Andere bekende typen, zoals Duration, Struct, en Any, worden afgewezen; codeer die waarden in plaats daarvan als een ondersteunde scalair of bericht. De fixed32 en fixed64 scalaire types en map met niet-string-toetsen worden ook niet ondersteund.

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

Data decoderen met schema's

Databricks decodeert elk bericht met de Spark's from_json, from_avro, en from_protobuf functies. De volgende gedragingen gelden of je het schema inline declareert of oplost vanuit een schemaregister:

• Misvormde dossiers. Decoding gebruikt de PERMISSIVE modus, dus een record dat niet overeenkomt met zijn schema decodeert naar een nullwaarde in plaats van de stroom te laten falen.
• Avro-vakbonden. Een unie van meerdere recordtypes decodeert naar een struct met één veld per recordtype, elk genoemd naar zijn Avro-record.
• Protubuf-typen. Unsigned gehele getallen decoderen naar een breder signed type (bijvoorbeeld StringValue naar Int32Value en DECIMAL(20,0) naar uint64), enumvelden decoderen naar hun naam als tekenreeks, en scalaire wrappertypen (bijvoorbeeld BIGINT en uint32) decoderen naar een nullable kolom van het ingepakte type.

Schemaregister

Schema-registries slaan schema's op en versies die streamingproducenten en consumenten gebruiken, en handhaven compatibiliteitsregels naarmate die schema's zich ontwikkelen. Wanneer een extern schemaregister is geconfigureerd, leest Feature Store het schema uit het register en gebruikt het om het streamingbericht te decoderen. Je declareert het schema niet inline op de Stream wanneer je een schema-register gebruikt.

Ondersteuning voor schema-registers kent de volgende beperkingen:

• Alleen ondersteund voor Kafka-streams.
• Alleen Confluent Schema Registry wordt ondersteund
• Alleen de Avro - en Protobuf-formaten worden ondersteund. Om JSON-berichten te lezen, definieer je in plaats daarvan het schema inline. Zie JSON-schema.
• Elke stroom is verbonden met precies één Confluent-subject voor de berichtwaarde, en één voor de berichtsleutel (indien beschikbaar). Stream topics met meerdere schemarecords zijn geen ondersteunde configuratie. Als je Stream verbinding maakt met onderwerpen die meerdere schema's bevatten, worden records die niet overeenkomen met het schema van het opgegeven onderwerp als null gedecodeerd.

Verbind met een schemaregister

Geef de registerverbindingsgegevens als opties op de Kafka Unity Catalog-verbinding en sla het register API-geheim op in een Databricks-geheim scope. De run-as-identiteit van de stream moet toestemming hebben READ op de geheime scope, omdat de ingestiepijplijn het geheim tijdens runtime leest. Voor hoe je een verbinding maakt en configureert, zie Create a connection.

Voeg de schema_registry_url, schema_registry_api_key, en schema_registry_api_secret opties toe aan de verbinding die wordt gebruikt voor authenticatie. Het volgende voorbeeld creëert een Kafka-verbinding die zich authenticeert bij de broker met een Unity Catalog-servicecredential en bij het register met een API-sleutel:

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>')
)

Stel zowel de optie schema_registry_api_secret in de Kafka-verbinding als de secret-scopeverwijzing in de Stream in op hetzelfde geheim.

Maak een stream aan die een schema-register gebruikt

Geef een SchemaRegistryConfig door als de schema_config. Verwijs naar het geheim van de registry API met api_secret_ref, en identificeer het onderwerp en het formaat met voor payload_schema_locator de berichtwaarde, of key_schema_locator voor de berichtsleutel. Er moet minstens één locatieaanduiding worden opgegeven.

Let op de verschillen hier vergeleken met de directe schemavoorbeelden in de sectie Schemaconfiguratie . Wanneer je een schemaregister gebruikt, geef je het schema niet inline door aan de Stream naar schema_config. In plaats daarvan specificeer je een SchemaRegistryConfig die het schema in het register identificeert.

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

Een Confluent-subject is de benoemde scope waaronder de versiegeschiedenis van een schema wordt geregistreerd en compatibiliteit wordt afgedwongen. Stel subject in op de naam van de relevante scope, die meestal wordt bepaald door de strategie voor onderwerpnamen:

• TopicNameStrategy (standaard, leidt het onderwerp af van de topicnaam): <topic>-value voor de waarde en <topic>-key voor de sleutel. Bijvoorbeeld, het waardeschema voor het onderwerp transactions gebruikt het onderwerp transactions-value.
• RecordNameStrategy (leidt het onderwerp af van de naam van het schema, onafhankelijk van het onderwerp): de volledig gekwalificeerde recordnaam, zoals com.example.Payment. Dit is de naamruimte en de naam van het record voor Avro, of het pakket en de naam van het bericht voor Protobuf.
• TopicRecordNameStrategy (combineert de onderwerp- en recordnamen): <topic>-<fully-qualified-record-name>, zoals transactions-com.example.Payment.

format is vereist. Stel het in op SchemaLocatorFormat.FORMAT_AVRO of SchemaLocatorFormat.FORMAT_PROTOBUF om te passen bij hoe het onderwerp geserialiseerd is.

Schema-evolutie

De ingestiepijplijn lost het huidige schema van het subject op zodra het begint. Wanneer je een nieuwe achterwaarts compatibele schemaversie registreert op het onderwerp in het schemaregister, blijft de lopende pijplijn de versie gebruiken waarmee het begon.

Voor streams die gebruikmaken van een schemaregister wordt de innamepipeline automatisch om de paar uur opnieuw gestart. Bij elke herstart wordt de nieuwste schemaversie van het onderwerp opgepikt, en verschijnen er nieuwe of gewijzigde velden in de invoertabel.

Stromen die directe schema's gebruiken in plaats van een schema-register evolueren hun schema met update_stream. Zie Update a stream.

Voor hoe de pijplijn records behandelt die niet overeenkomen met het schema dat hij momenteel gebruikt, zie Data Decoding met schema's.

Filter records op type

Een Stream decodeert elk record met één sleutel- en waardeschema (indien beschikbaar), of je ze nu direct specificeert of een schemaregister gebruikt. Omdat een onderwerp meer dan één type record kan dragen en Streams zich op meerdere onderwerpen kunnen abonneren, kun je record_type_filter selecteren welke records uit het onderwerp tot deze Stream behoren.

Bied een SQL-expressie die verwijst naar gedecodeerde velden met puntnotatie, bijvoorbeeld value.event_type = 'transaction'. Records die niet aan het filter voldoen, worden genegeerd. Ze worden niet in de innametabel geschreven en worden niet gebruikt bij materialisatie. Om een stream voor andere recordtypes te maken, maak je een aparte stream aan met een andere record_type_filter.

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    # ...source, connection, schema, and ingestion config...
    record_type_filter="value.event_type = 'transaction'",
)

Zelfs zonder record_type_filter, faalt decoderen nooit de Stream. Een record dat niet overeenkomt met het geconfigureerde schema wordt permissief gedecodeerd. Records worden op een van de volgende manieren gedecodeerd:

• In een rij met NULL waarden voor de velden die het schema verwacht, maar die het record weglaat (JSON, Avro en Protobuf).
• Naar een rij met waarden die tot een ander recordtype behoren (alleen Avro en Protobuf).

Om te bepalen of een rij in de invoertabel tot het verwachte recordtype behoort, gebruik je een van de volgende controles:

• Controleer bijvoorbeeld of een veld gelijk is aan een verwachtingswaarde value.event_type = 'transaction' (voorkeur voor Avro en Protobuf).
• Controleer bijvoorbeeld of een veld niet-isNULL.value.activity_id IS NOT NULL

Het wordt aanbevolen om met aparte streams te werken record_type_filter wanneer de schema's aanzienlijk verschillen tussen recordtypes op het onderwerp of je de toegang tot elk recordtype onafhankelijk wilt regelen. Om de kosten te behouden, raadt Databricks aan om een klein aantal Streams te behouden, aangezien elke Stream een aparte innamepijplijn en innametabel heeft. Elke stroom gebruikt ook aparte rekenkracht tijdens materialisatie. Je kunt feature-specifieke filters gebruiken voor materialisatie.

record_type_filter verschilt van de filter_condition van een functie. record_type_filter wordt ingesteld op de Stream en bepaalt welke records worden ingevoerd en beschikbaar zijn voor alle features die de Stream als bron gebruiken, terwijl filter_condition deze op een individuele feature wordt ingesteld en rijen filtert vóór aggregatie. Zie Filtercondities op streamingbronnen voor meer details over filter_condition.

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 stroom wordt aangemaakt, start Databricks een beheerde ingestiepijplijn die continu berichten uit de bronstream leest en deze in een Delta-tabel (de ingestie-tabel) schrijft. De pijplijn start vanaf de laatste positie in de bron en draait continu, waarbij alleen nieuwe berichten worden verwerkt 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 invoertabel bevat de berichtgegevens samen met metadatakolommen. De gemeenschappelijke kolommen zijn aanwezig voor elke bron; de kafka_* kolommen zijn alleen aanwezig voor een Kafka-stroom.

Bron voor terugvullen

Omdat de forward-fill-pipeline begint vanaf de meest recente positie in de bron, worden berichten die al bestonden voordat de stream werd aangemaakt niet meegenomen. 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 metadatakolommen worden doorgegeven als ze in de backfillbron aanwezig zijn, of anders op NULL ingesteld. Voor Kafka zijn dit kafka_topic, kafka_partition en 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"
    ),
)

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 ingestiepijplijn begint vanaf de laatste positie in de bron, vangt deze alleen berichten op die binnenkomen nadat de stroom is aangemaakt. 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"],
)

Kostentoeschrijving

Stel tags en budget_policy_id in voor de IngestionConfig om de kosten van de beheerde gegevensinname van de stream toe te wijzen. Azure Databricks past deze toe op de Lakeflow-opnamepijplijn en de forward-fill- en backfill-taken wanneer de Stream wordt aangemaakt.

Zie Kosten toeschrijven met tags en serverless-gebruiksbeleid voor een voorbeeld van de taglimieten en hoe je de toegeschreven uitgaven kunt opvragen.

Kolommen uitsluiten uit een stroom

Gebruik excluded_columns het om specifieke kolommen uit een stream te verwijderen die je niet wilt invoeren. Een uitgesloten kolom wordt niet naar de invoertabel geschreven en kan niet door een functie worden verwezen of in training worden gebruikt.

Specificeer elke kolom met behulp van dotnotatie in de berichtsleutel of waarde, zoals value.user.email of key.account_id. Deze kolommen worden verwijderd uit de gedecodeerde key en value bij inname, backfill en materialisatie. Als een pad naar een struct wijst, worden ook al zijn geneste velden verwijderd (bijvoorbeeld, value.address ook vallen value.address.city en value.address.zip).

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    # ...source, connection, schema, and ingestion config...
    excluded_columns=["value.user.email", "value.user.ssn"],
)

Bij gebruik van directe schema's moet de uitgesloten kolom al bestaan in het sleutel- of waardeschema, anders create_stream faalt het. Wanneer je een schemaregister gebruikt, kun je een kolom uitsluiten voordat die bestaat. Een uitgesloten kolom kan ook geen deduplicatiekolom zijn, omdat deduplicatiekolommen nodig zijn om dubbele rijen te identificeren. Elke functie die verwijst naar een uitgesloten kolom (bijvoorbeeld als entiteit, tijdreeks of invoer) wordt niet aangemaakt.

Je kunt de uitgesloten kolommen van een Stream na het aanmaken wijzigen met update_stream, zowel voor Streams met een direct schema als voor Streams die gebruikmaken van Schema Registry. Zie Update a stream voor meer details.

Beheer van stromen

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.

Werk een stream bij

Gebruik update_stream om een stream na het aanmaken te wijzigen. Pass schema_config om een direct schema te ontwikkelen, excluded_columns om te veranderen welke kolommen worden weggelaten, of beide. Het bijwerken van andere velden wordt niet ondersteund. Maak in plaats daarvan een nieuwe stream aan.

Het updaten van een stroom start de invoerpijplijn opnieuw zodat de wijziging in werking treedt. De inname wordt meestal binnen enkele minuten hervat.

Ontwikkel een direct schema

Voor een stroom die directe schema's gebruikt, geef een door aan DirectSchemasschema_config. Stel payload_schema, key_schema, of beide. Een kant die je niet kiest, blijft ongewijzigd. Schema-registerondersteunde streams weigeren een schema_config update en moeten in plaats daarvan via het register worden ontwikkeld.

from databricks.feature_engineering.entities import DirectSchemas, SchemaConfig

stream = client.update_stream(
    name="my_catalog.my_schema.my_stream",
    schema_config=DirectSchemas(
        payload_schema=SchemaConfig(
            json_schema=(
                '{'
                '  "type": "object",'
                '  "properties": {'
                '    "user_id": {"type": "string"},'
                '    "amount": {"type": "number"},'
                '    "event_time": {"type": "string"},'
                '    "channel": {"type": "string"}'
                '  }'
                '}'
            )
        ),
    ),
)

Schema-updates moeten achterwaarts compatibel zijn zodat de draaiende innamepijplijn bestaande records kan blijven decoderen en naar de innametabel kan schrijven. Alle andere wijzigingen worden afgewezen.

Wat is toegestaan hangt af van het formaat:

• JSON en Protobuf: voeg optionele velden toe, verwijder velden en verbreed het type van een veld (bijvoorbeeld int naar bigint). Protobuf maakt ook het herschikken van velden mogelijk.
• Avro: staat alleen verbreden int naar long en het verwijderen van een achterloopveld waarvan de bytes geen later veld wordt gelezen. Om een Avro-schema vrijer te ontwikkelen, gebruik je in plaats daarvan een schema-registratiestream.

Het toevoegen van velden laat de codering key en value structs van de ingestietabel groeien. Rijen die vóór de update zijn geschreven, behouden hun oorspronkelijke vorm, en de toegevoegde velden lezen als NULL voor die eerdere rijen. Verwijderingen en typewijzigingen gelden alleen voor records die na de update zijn ingevoerd.

Uitgesloten kolommen wijzigen

Geef de volledige nieuwe set kolompaden door naar excluded_columns, die de bestaande set vervangen. Geef een lege lijst ([]) om alle uitsluitingen te verwijderen. Voor details over dit gedrag, zie Kolommen uitsluiten van een stream.

stream = client.update_stream(
    name="my_catalog.my_schema.my_stream",
    excluded_columns=["value.user.email", "value.user.ssn"],
)

Het wijzigen van uitgesloten kolommen is alleen vooruit. Nieuw uitgesloten kolommen worden niet meer geschreven (verschijnen op NULL) en nieuw toegevoegde kolommen worden vanaf het begin ingevuld ingevuld terwijl eerder geschreven rijen as-is. Om te voorkomen dat er ooit een nieuwe kolom wordt ingenomen:

• Schemaregister: voeg de kolom eerst toe aan excluded_columns en wacht tot de innamepijplijn opnieuw is gestart, en registreer vervolgens de nieuwe schemaversie in het register.
• Directe schema's: voeg de kolom toe aan schema_config en aan excluded_columns in dezelfde update_stream aanroep.

Een stream verwijderen

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

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