Konfigurera en dataström

En ström representerar en extern strömmande datakälla, till exempel Apache Kafka. Strömmar lagrar anslutningsinformation, autentisering, scheman och inmatningskonfiguration. När en ström har skapats kan du referera till den med hjälp av funktionsvydefinitioner för att skapa realtidsströmningsfunktioner.

Strömmar har tredelade namn (catalog.schema.stream_name). Åtkomst till en Stream styrs av dess associerade inmatningstabell. Se Intag och fyllning för detaljer.

Requirements

• För att köra notebook-kommandon: serverlöst eller ett klassiskt beräkningskluster som kör Databricks Runtime 17.0 ML eller senare.
• feature-engineering-client Python-paketet version 0.18.0 eller senare måste installeras.

Anslutning till strömkällor

Innan du definierar strömningsfunktioner ansluter och testar du en strömmande Lakeflow-pipelineanslutning till din Kafka-mäklare. Feature Store bygger på serverless SDP, vilket innebär att du behöver en mekanism för att koppla din klassiska beräkning (mäklare eller endpoint) till Databricks serverlösa beräkning. Detta görs via produkter som privatelink eller genom att tillåta din klassiska beräkning att vara tillgänglig från det publika internet.

Skapa en strömning

Använd create_stream() för att skapa en ny Stream. En Stream kräver fyra konfigurationskomponenter:

• Källkonfiguration: Specificerar streamingplattformen och källkodsspecifika detaljer, såsom ämnesprenumeration för en Kafka-källa.
• Anslutningskonfiguration: Anger hur du ansluter och autentiserar till strömningsplattformen, inklusive bootstrap-servrar och autentiseringsuppgifter.
• Schemakonfiguration: Definierar strukturen för meddelandenycklar och -värden.
• Inmatningskonfiguration: Anger var och hur dataström matas in. Se Intag och fyllning för detaljer.

För källkodsspecifik source_config och anslutningsuppsättning, tillsammans med ett komplett create_stream() exempel, se Apache Kafka. Alternativen för schema och inmatning är gemensamma för alla källor.

Apache Kafka

För att strömma från Apache Kafka, använd KafkaStreamConfig som källkonfiguration och en Unity Catalog-anslutning för autentisering. Se Streaming i serverlös beräkningsmiljö och Ansluta till Apache Kafka om Kafka-anslutning.

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-prenumerationslägen

Prenumerationsläget anger hur Stream väljer Kafka-ämnen som den ska konsumera från. Tre lägen stöds:

Kafka-autentisering

Anslutning till Unity Catalog (rekommenderas)

Använd en Unity Catalog-anslutning för att autentisera till ditt Kafka-kluster. Det här är den rekommenderade metoden för hanterad autentisering. Information om hur du skapar en anslutning finns i Skapa en anslutning. Streamens skapare måste ha USE CONNECTION anslutningen aktiverad. Alla användare som materialiserar funktioner med strömmen som källa måste också ha USE CONNECTION på anslutningen.

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

Anslutningen stöder både IAM (tjänstelegitimation) och SASL-autentisering.

IAM (tjänstelegitimation)

Autentisera med en Unity Catalog-tjänstelegitimation, till exempel för att ansluta till Amazon MSK med IAM. Information om hur du skapar en tjänstautentiseringsuppgift finns i Skapa autentiseringsuppgifter för tjänsten. Ställ in tjänstelegitimationsnamnet med valet credential :

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

Utöver USE CONNECTION på anslutningen behöver identiteter som använder tjänstelegitimationen även ACCESS på anslutningen. Lämna ACCESS den refererade tjänstelegitimationen till streamens skapare och till alla identiteter som ger funktioner med streamen. Se Bevilja behörighet för att använda en tjänstebaserad autentiseringsuppgift för att få tillgång till en extern molntjänst.

SASL

SASL-autentisering använder användarnamn och lösenord. Ange sasl_mechanism till något av följande:

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

Ange autentiseringsuppgifterna med alternativen password och user. Anslutningen lagrar dessa uppgifter säkert.

Följande exempel använder SASL/SCRAM. För SASL/PLAIN, sätt sasl_mechanism till 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>'
)

Direkt mTLS

För direkt mTLS-autentisering anger du keystore- och truststore-filer som lagras på en Unity Catalog-volym, med lösenord som anges via Databricks secret scopes. Mer information om SSL-autentisering med Kafka finns i Använda SSL för att ansluta Azure Databricks till 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"
        ),
    ),
)

Schemakonfiguration

Definiera strukturen för meddelandenycklar och värden så att inläsning och funktionsdefinitioner kan läsa enskilda fält. För Kafka-källor payload_schema motsvarar det Kafka-meddelandevärdet ( value i Kafkas nyckelvärdesmodell) och key_schema motsvarar Kafka-meddelandenyckeln. Minst en av payload_schema eller key_schema måste tillhandahållas.

Varje SchemaConfig format accepterar ett av tre format, som matchar hur källan serialiserar sina meddelanden: json_schema, avro_schema, eller proto_schema. Om inget schema anges för en nyckel eller nyttolast behandlas det som en enkel sträng.

Kodexemplen i detta avsnitt använder scheman deklarerade inline med DirectSchemas, där schemat tillhandahålls som en sträng. För att hantera scheman med ett externt schemaregister, se Schema-register för detaljer.

JSON-schema

Ange en JSON-schemasträng till 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

Ange en Avro-schemasträng till avro_schema. Avro-logiska typer stöds, inklusive timestamp-millis, date, och 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

Ange en ProtoSchemaSpec till proto_schema med Protocol Buffers.proto-källtexten och namnet på nyttolastmeddelandet. Importera ProtoSchemaSpec från databricks.feature_engineering.entities.

message_name måste vara det fullt kvalificerade meddelandenamnet, inklusive det package deklarerade i texten .proto (till exempel com.example.Event, inte Event). Både proto2- och proto3-syntax stöds.

google.protobuf.Timestamp och skalär-wrappertyperna (StringValue, Int32Value, och så vidare) stöds, och deras importer löses automatiskt. Andra välkända typer, såsom Duration, Struct, och Any, avvisas; koda istället dessa värden som en stödd skalär eller meddelande. Och fixed64 skalärtyperna fixed32 och map med icke-strängnycklar stöds inte heller.

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

Avkodning av data med hjälp av scheman

Databricks avkodar varje meddelande med Sparks from_json, from_avro, och from_protobuf funktioner. Följande beteenden gäller oavsett om du deklarerar schemat inline eller löser det från ett schemaregister:

• Felaktigt formaterade poster. Avkodning använder läget PERMISSIVE, så en post som inte matchar sitt schema avkodas till ett nullvärde i stället för att strömmen avbryts.
• Avro-fackföreningar. En union med flera posttyper avkodas till en struktur med ett fält per posttyp, där varje fält är uppkallat efter motsvarande Avro-post.
• Protobuf-typer. Osignerade heltal avkodar till en bredare signerad typ (till exempel uint32 till BIGINT och uint64 till DECIMAL(20,0)), enumfält avkodar till sitt strängnamn, och skalär-wrappertyper (till exempel StringValue och Int32Value) avkodas till en nullbar kolumn av den wrappade typen.

Schemaregister

Schemaregister lagrar och versionerar scheman som streamingproducenter och konsumenter använder, och upprätthåller kompatibilitetsregler i takt med att dessa scheman utvecklas. När ett externt schemaregister konfigureras läser Feature Store schemat från registret och använder det för att avkoda strömningsmeddelandet. Du deklarerar inte schemat direkt i dataströmmen när du använder ett schemaregister.

Stöd för schemaregister har följande begränsningar:

• Stöds endast för Kafka-strömmar.
• Endast Confluent Schema Registry stöds
• Endast Avro - och Protobuf-formaten stöds. För att läsa JSON-meddelanden, deklarera schemat som en inline istället. Se JSON-schemat.
• Varje ström är kopplad till exakt ett Confluent-subjekt för meddelandevärdet och ett för meddelandenyckeln (om det finns). Strömteman som innehåller flera schemaposter är inte en stödd konfiguration. Om din Stream är ansluten till topics som innehåller flera scheman avkodas poster som inte överensstämmer med schemat för det angivna subjektet som null.

Koppla till ett schemaregister

Ange registeranslutningsdetaljer som alternativ på Kafka Unity Catalog-anslutningen och lagra register-API:s hemlighet i ett Databricks-hemlighetsområde. Streamens run-as-identitet måste ha READ tillstånd på hemlighetsområdet, eftersom inmatningspipelinen läser hemligheten vid körning. För hur man skapar och konfigurerar en anslutning, se Skapa en anslutning.

Lägg till schema_registry_url, schema_registry_api_key, och schema_registry_api_secret alternativen till anslutningen som används för autentisering. Följande exempel skapar en Kafka-anslutning som autentiseras till mäklaren med en Unity Catalog-tjänstelegitimation och till registret med en API-nyckel:

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

Ställ in både alternativet schema_registry_api_secret i Kafka-anslutningen och referensen till den hemliga omfattningen på streamen till samma hemlighet.

Skapa en ström som använder ett schema-register

Skicka SchemaRegistryConfig som schema_config. Referera register-API-hemligheten med api_secret_ref, och identifiera ämnet och formatet med payload_schema_locator för meddelandevärdet, eller key_schema_locator för meddelandenyckeln. Minst en locator måste tillhandahållas.

Notera skillnaderna här jämfört med exemplen med direkta scheman i avsnittet Schemakonfiguration . När du använder ett schemaregister tillhandahåller du inte schemat infogat direkt i strömmen till schema_config. Istället specificerar du en SchemaRegistryConfig som identifierar schemat i registret.

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

Ett Confluent-subjekt är det namngivna omfång under vilket ett schemas versionshistorik registreras och kompatibilitet upprätthålls. Sätt subject till det relevanta scope-namnet, vilket vanligtvis bestäms från ämnesnamnstrategin:

• TopicNameStrategy (standard, härleder subjektet från ämnesnamnet): <topic>-value för värdet och <topic>-key för nyckeln. Till exempel använder värdeschemat för ämnet transactions subjektet transactions-value.
• RecordNameStrategy (härleder subjektet från schemats postnamn, oberoende av ämnet): det fullt kvalificerade postnamnet, såsom com.example.Payment. Detta är postens namnrymd och namn för Avro, eller meddelandets paket och namn för Protobuf.
• TopicRecordNameStrategy (kombinerar ämnes- och postnamn): <topic>-<fully-qualified-record-name>, såsom transactions-com.example.Payment.

format måste anges. Ställ in den till SchemaLocatorFormat.FORMAT_AVRO eller SchemaLocatorFormat.FORMAT_PROTOBUF för att matcha hur ämnet är serialiserat.

Schemautveckling

Inmatningspipelinen löser upp subjektets aktuella schema när det startar. När du registrerar en ny bakåtkompatibel schemaversion på subjektet i schemaregistret fortsätter den körande pipelinen att använda den version den började med.

För schema-registerstödda strömmar startar inmatningspipelinen automatiskt om med några timmars mellanrum. Vid varje omstart plockar den upp subjektets senaste schemaversion, och nya eller ändrade fält visas i inmatningstabellen.

Strömmar som använder direktscheman i stället för ett schemaregister utvecklas med update_stream. Se Uppdatera en dataström.

För hur pipelinen hanterar poster som inte matchar det schema den använder just nu, se Decoding data using schemas.

Filtrera poster efter typ

En ström avkodar varje post med ett enda nyckel- och värdeschema (om det finns), oavsett om du specificerar dem direkt eller använder ett schema-register. Eftersom ett topic kan innehålla mer än en typ av datapost och Streams kan prenumerera på flera topics, använd record_type_filter för att välja vilka dataposter från topicen som tillhör den här Streamen.

Tillhandahåll ett SQL-uttryck som refererar till avkodade fält med punktnotation, till exempel value.event_type = 'transaction'. Poster som inte matchar filtret ignoreras. De skrivs inte till inmatningstabellen och används inte vid materialisering. För att skapa en ström för andra posttyper, skapa en separat ström med en annan 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'",
)

Även utan record_type_filter orsakar avkodningen aldrig att Stream misslyckas. En post som inte matchar det konfigurerade schemat avkodas tillåtande. Posterna avkodas på ett av följande sätt:

• Till en rad med värdena NULL för de fält som schemat förväntar sig men som posten saknar (JSON, Avro och Protobuf).
• Till en rad med värden som tillhör en annan posttyp (endast Avro och Protobuf).

För att identifiera om en rad i inmatningstabellen tillhör den förväntade posttypen, använd en av följande kontroller:

• Kontrollera att ett fält motsvarar ett förväntat värde, till exempel value.event_type = 'transaction' (rekommenderas för Avro och Protobuf).
• Kontrollera att ett fält inte är NULL, till exempel value.activity_id IS NOT NULL.

Det rekommenderas att använda record_type_filter med separata Streams när schemana skiljer sig avsevärt mellan posttyper i ämnet eller när du vill styra åtkomsten till varje posttyp oberoende av varandra. För att hålla kostnaderna nere rekommenderar Databricks att du har ett litet antal Streams, eftersom varje Stream har en separat inmatningspipeline och inmatningstabell. Varje Stream använder också separata beräkningsresurser vid materialisering. Du kan använda funktionsspecifika filter för materialisering.

record_type_filter är annorlunda än en egenskaps filter_condition. record_type_filter sätts på strömmen och styr vilka poster som tas in och är tillgängliga för alla funktioner som använder strömmen som källa, medan filter_condition den sätts på en enskild funktion och filtrerar rader före aggregering. Se Filtervillkor för strömningskällor för mer information om filter_condition.

Intag och återfyllning

Parametern ingestion_config konfigurerar hur dataströmdata samlas in och lagras för träning och servering.

Åtkomst till en stream styrs av inmatningstabellen:

• SELECT i inmatningstabellen ger läsåtkomst till Stream.
• MANAGE i inmatningstabellen ger borttagningsåtkomst.

Mer information om tabellbehörigheter finns under Tabell och referens för behörigheter i Unity Catalog.

Inmatningspipeline

När en ström skapas startar Databricks en hanterad intagningspipeline som kontinuerligt läser meddelanden från källströmmen och skriver dem till en Delta-tabell (intagningstabellen). Pipelinen startar från den senaste positionen i källkoden och körs kontinuerligt, och fångar endast nya meddelanden som anländer efter att strömmen har skapats. Den här inmatningstabellen används för träning med strömningsfunktioner. När en dataström tas bort tas även inmatningspipelinen och inmatningstabellen bort.

Inmatningsmål

ingestion_destination Anger det tredelade Delta-tabellnamnet där dataströmmen skrivs.

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

Tabellschema för inmatning

Insamlingstabellen innehåller meddelandedata tillsammans med metadatakolumner. De gemensamma kolumnerna finns för varje källa; kolumnerna kafka_* finns endast för en Kafka-ström.

Återfyllnadskälla

Eftersom forward-fill-pipelinen börjar från den senaste positionen i källkoden fångar den inte upp meddelanden som fanns innan strömmen skapades. Konfigurera en valfri återfyllnadskälla för att tillhandahålla historisk datatäckning för träning.

När en återfyllnadskälla har konfigurerats kör Databricks ett engångsjobb MERGE INTO som kopierar återfyllnadsrader till inmatningstabellen med record_source="backfill". MERGE körs först efter att överlappningskontrollen bekräftar att återfyllnadskällan och dataströmmen för vidarebefordran har överlappande tidsstämplar (se Överlappning mellan återfyllnads- och liveströmdata). Om överlappningsvillkoret inte uppfylls inom 2 dagar körs MERGE ändå för att undvika blockering på obestämd tid.

Tabellen för återfyllnad måste innehålla en stream_record_timestamp kolumn av typen TIMESTAMP i UTC-tidszonen. Andra metadatakolumner skickas vidare om de finns i backfillkällan, annars anges de som NULL. För Kafka är dessa kafka_topic, kafka_partition och 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"
    ),
)

Överlappning mellan återfyllnads- och liveströmdata

Innan du kör en MERGE mellan återfyllningen och inmatningstabellen jämför en överlappningskontroll tidsstämplarna i de två tabellerna:

• Max för återfyllnad: Maximalt stream_record_timestamp i återfyllnadskällan.
• Inmatningsmin: Det minsta stream_record_timestamp antalet rader (record_source="stream") i inmatningstabellen.

MERGE fortsätter när återfyllningens senaste tidsstämpel överskrider inmatningstabellens tidigaste tidsstämpel med minst 1 timme. Den här överlappningen säkerställer att det inte finns några luckor i inmatningstabellen. Om överlappningsvillkoret inte uppfylls inom 2 dagar körs MERGE ändå för att undvika blockering på obestämd tid.

Eftersom inmatningspipelinen startar från den senaste positionen i källan, fångar den endast meddelanden som anländer efter att strömmen har skapats. Din återfyllnadskälla måste innehålla data som sträcker sig in i intagningstidsintervallet – inte bara fram till tidpunkten då strömmen skapades.

Om du till exempel skapar en ström kl. 15:00 börjar pipelinen för vidarebefordran att läsa meddelanden från 15:00 och framåt. Din återfyllnadskälla måste innehålla data med tidsstämplar till minst 16:00 (1 timme efter start av framåtfyllning) för att uppfylla överlappningskontrollen. Det innebär att du bör uppdatera din återfyllnadstabell efter 16:00 för att säkerställa att inmatningstabellen inte har några luckor.

Deduplication

Använd deduplication_columns för att ange kolumnsökvägar för att identifiera duplicerade rader vid inmatning av backfill-data och strömmande forward-fill-data. Använd punkt notation för kapslade fält (till exempel "value.user_id").

Välj dedupliceringskolumner baserat på dina data:

• Om varje post i dataströmmen innehåller en unik identifierare (till exempel value.transaction_id), använder du den kolumnen för deduplicering.
• Om din återfyllnadskälla innehåller kafka_partition och kafka_offset kolumner använder du dem för att unikt identifiera varje post.
• Om inga dedupliceringskolumner anges är standarddedupliceringsnyckeln den fullständiga kombinationen av key, valueoch stream_record_timestamp. Detta rekommenderas inte eftersom den här strikta villkorsmatchningen enkelt kan leda till dubbletter.

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

Kostnadstillskrivning

Sätt tags och budget_policy_id på IngestionConfig för att attribuera kostnaden för Streamens hanterade inmatning. Azure Databricks tillämpar dem på insamlings-Lakeflow-pipelinen och dess forward-fill- och backfill-jobb när strömmen skapas.

För exempel på tagggränser och hur du frågar efter den attribuerade kostnaden, se Attribuera kostnader med taggar och användningsprinciper för serverlösa tjänster.

Exkludera kolumner från en ström

Använd excluded_columns det för att ta bort specifika kolumner från en ström som du inte vill ta in. En utesluten kolumn skrivs inte till insamlingstabellen och kan inte refereras av någon funktion eller användas i träning.

Ange varje kolumn med punktnotation i meddelandets nyckel eller värde, till exempel value.user.email eller key.account_id. Dessa kolumner tas bort från de avkodade key och value i samband med inläsning, backfill och materialisering. Om en väg pekar på en struktur tas alla dess nästlade fält också bort (till exempel value.address släpper value.address.city även och 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"],
)

När direkta scheman används måste den uteslutna kolumnen redan finnas i nyckel- eller värdeschemat, annars misslyckas create_stream. När du använder ett schema-register kan du utesluta en kolumn innan den existerar. En utesluten kolumn kan inte heller vara en dedupliceringskolumn, eftersom dedupliceringskolumner krävs för att identifiera dubblettrader. En funktion som refererar till en exkluderad kolumn (till exempel som entitet, tidsserie eller indata) går inte att skapa.

Du kan ändra en ströms uteslutna kolumner efter skapandet med update_stream, både på direkt schema och schema-registerstödda strömmar. Se Uppdatera en ström för mer information.

Hantera vattendrag

Hämta en dataström

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

Visa strömmar

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

Ange include_schemas=True för att inkludera fullständig schemainformation. Scheman kan vara stora och detta kan resultera i en tidskrävande åtgärd. Om du vill hämta scheman individuellt i stället använder du get_stream.

Uppdatera en ström

Använd update_stream för att ändra en Stream när den har skapats. pass schema_config för att utveckla ett direkt schema, excluded_columns för att ändra vilka kolumner som tas bort, eller båda. Det går inte att uppdatera andra fält. Skapa istället en ny ström.

Att uppdatera en ström startar om dess intagningspipeline så att ändringen träder i kraft. Intaget återupptas vanligtvis inom några minuter.

Skapa en direkt schemadefinition

För en ström som använder direkta scheman, skicka a DirectSchemas till schema_config. Sätt payload_schema, key_schema, eller båda. En sida du inte bestämmer är oförändrad. Schema-registerstödda strömmar avvisar en schema_config uppdatering och måste istället utvecklas genom registret.

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

Schemauppdateringar måste vara bakåtkompatibla så att den körande inmatningspipelinen kan fortsätta avkoda befintliga poster och skriva till inmatningstabellen. Andra ändringar avvisas.

Vad som är tillåtet beror på formatet:

• JSON och Protobuf: lägg till valfria fält, ta bort fält och bredda ett fälts typ (till exempel int till bigint). Protobuf tillåter också omordning av fält.
• Avro: tillåter endast utvidgning int till long och borttagning av ett efterföljande fält vars bytes inget senare fält läses. För att utveckla ett Avro-schema mer fritt, använd istället en schema-registerstödd ström.

Att lägga till fält växer intagningstabellens avkodning key och value strukturer. Rader skrivna före uppdateringen behåller sin ursprungliga form, och de tillagda fälten visas som NULL för de tidigare raderna. Borttagningar och typändringar träder endast i kraft för poster som matas in efter uppdateringen.

Ändra uteslutna kolumner

Skicka den fullständiga nya uppsättningen kolumnvägar till excluded_columns, vilket ersätter den befintliga mängden. Skicka en tom lista ([]) för att rensa alla undantag. För detaljer om detta beteende, se Uteslut kolumner från en ström.

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

Ändringar av uteslutna kolumner kan endast göras framåt. Nyligen uteslutna kolumner slutar skrivas (visas vid NULL) och nyligen inkluderade kolumner börjar fyllas framöver, medan tidigare skrivna rader lämnas as-is. För att förhindra att en ny kolumn någonsin intöss:

• Schema-register: lägg till kolumnen först excluded_columns och vänta på att inmatningspipelinen startar om, registrera sedan den nya schemaversionen i registret.
• Direkta scheman: lägg till kolumnen till schema_config och till excluded_columns i samma update_stream anrop.

Ta bort en dataström

Att ta bort en dataström tar även bort både dess inmatningspipeline och inmatningstabell.

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

Exempelanteckningsbok

Ett heltäckande exempel som skapar en Stream, definierar streamingfunktioner och driftsätter till en slutpunkt för servering finns i följande notebook:

Snabbstartsguide för strömmande funktionsvyer

Hämta anteckningsbok