Nastavení streamu

Důležité

Tato funkce je ve verzi Public Preview. Správci pracovního prostoru můžou řídit přístup k této funkci ze stránky Previews . Viz Manage Azure Databricks preview.

Stream představuje externí streamový zdroj dat, například Apache Kafka. Streamy ukládají podrobnosti o připojení, ověřování, schémata a konfiguraci příjmu dat. Po vytvoření streamu na něj můžete odkazovat pomocí definic zobrazení funkcí a vytvářet funkce streamování v reálném čase.

Streamy mají názvy tří částí (catalog.schema.stream_name). Přístup ke streamu se řídí přidruženou tabulkou příjmu dat. Podrobnosti najdete v části Ingestování a zpětné doplnění dat.

Požadavky

Pro spouštění příkazů poznámkového bloku: bezserverový nebo klasický výpočetní cluster se spuštěným modulem Databricks Runtime 17.0 ML nebo novějším.
Musí feature-engineering-client být nainstalován balíček Python verze 0.16.0 nebo vyšší.

Vytvořte stream

Slouží create_stream() k vytvoření nového streamu. Stream vyžaduje čtyři součásti konfigurace:

Konfigurace zdroje: Určuje platformu streamování (například Kafka) a podrobnosti specifické pro zdroj (například odběr témat pro Kafka).
Konfigurace připojení: Určuje, jak se připojit a ověřit na platformě streamování, včetně serverů bootstrap a přihlašovacích údajů.
Konfigurace schématu: Definuje strukturu klíčů a hodnot zpráv.
Konfigurace příjmu dat: Určuje, kde a jak se ingestují data streamu. Podrobnosti najdete v části Ingestování a zpětné doplnění dat.

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

Připojení ke zdrojům streamu

Před definováním funkcí streamování se připojte a otestujte připojení kanálu Stream Lakeflow k vašemu zprostředkovateli Kafka. Viz Streamování na výpočetních prostředcích bez serveru a připojení k Apache Kafka.

Informace o spravovaném streamování AWS (Amazon MSK) najdete v tématu Bezserverové privátní připojení k Amazon MSK. Podrobnosti o možnostech ověřování Kafka najdete v tématu Ověřování.

Authentication

Připojení katalogu Unity (doporučeno)

K autentizaci vůči clusteru Kafka použijte připojení Unity Catalog. Tento přístup se doporučuje pro spravované ověřování. Pokud chcete vytvořit připojení, přečtěte si téma Vytvoření připojení.

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

Přímé mTLS

Pro přímé ověřování mTLS zadejte soubory keystore a truststore uložené ve svazku Unity Catalog, přičemž hesla jsou odkazována prostřednictvím oblastí secret scope v Databricks. Další informace o ověřování SSL v systému Kafka najdete v tématu Použití protokolu SSL pro připojení Azure Databricks k Systému 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

Ověřování SASL (SASL/SCRAM i SASL/PLAIN) se během verze Preview nepodporuje.

Režimy předplatného

Režim odběru určuje, jak Stream vybírá témata Kafka, ze kterých bude číst. Podporují se tři režimy:

Mode	Description	Příklad
`subscribe`	Čárkami oddělený seznam názvů témat	`KafkaSubscriptionMode(subscribe="topic1,topic2")`
`subscribe_pattern`	Vzor regulárního výrazu v Javě pro porovnávání názvů témat	`KafkaSubscriptionMode(subscribe_pattern="events-.*")`
`assign`	JSON určující přiřazení témat-oddílů	`KafkaSubscriptionMode(assign='{"my-topic": [0, 1, 2]}')`

Konfigurace schématu

Definujte strukturu klíčů a hodnot zpráv pomocí formátu schématu JSON . U zdrojů Kafka payload_schema odpovídá hodnotě zprávy Kafka (value v modelu klíč–hodnota v systému Kafka) a key_schema odpovídá klíči zprávy Kafka. Musí být zadáno alespoň jedno z payload_schema nebo key_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"}'
    ),
)

Pokud pro klíč nebo datovou část není k dispozici žádné schéma, považuje se za jednoduchý řetězec.

Ingestace a zpětné doplnění dat

Parametr ingestion_config určuje, jak se data ze streamu zachycují a ukládají pro trénování a nasazení.

Přístup ke streamu se řídí tabulkou příjmu dat:

SELECT v tabulce příjmu dat udělí streamu přístup pro čtení.
MANAGE v tabulce příjmu dat uděluje přístup k odstranění.

Další informace o oprávněních k tabulkám najdete v tématech Tabulka a Referenční příručka k oprávněním v katalogu Unity Catalog.

Kanál příjmu dat

Když se vytvoří datový proud, Databricks spustí spravovaný kanál příjmu dat, který průběžně čte zprávy z tématu Kafka a zapisuje je do tabulky Delta (tabulka příjmu dat). Pipeline začíná na nejnovějším offsetu v systému Kafka a běží nepřetržitě, přičemž zachycuje pouze nové zprávy, které přijdou po vytvoření streamu. Tato tabulka příjmu dat se používá pro trénování s funkcemi streamování. Po odstranění datového proudu se odstraní také její kanál příjmu dat a tabulka příjmu dat.

Cíl ingestace

ingestion_destination určuje třídílný název tabulky Delta, do které se zapisují data streamu.

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

Schéma tabulky příjmu dat

Tabulka příjmu dat obsahuje data zpráv spolu se sloupci metadat:

Column	Typ	Description
`key`	Liší se (od `key_schema`)	Klíč zprávy Kafka strukturovaný podle schématu, které jste zadali.
`value`	Liší se (od `payload_schema`)	Hodnota zprávy Kafka (datový obsah), strukturovaná podle schématu, které jste poskytli.
`stream_record_timestamp`	`TIMESTAMP`	Časové razítko záznamu. U dat typu forward-fill jde o časové razítko příjmu brokeru Kafka. V případě zpětně doplňovaných dat jsou tato data dodaná zákazníkem.
`kafka_topic`	`STRING`	Téma Kafka, ze které byl záznam spotřebován.
`kafka_partition`	`INT`	Partice Kafka, ze které byl záznam načten.
`kafka_offset`	`LONG`	Posun Kafka záznamu v rámci jeho oddílu
`record_source`	`STRING`	Buď `"stream"` (doplnění z živého streamu Kafka), nebo `"backfill"` (ze zdroje zpětného doplnění).

Zdroj zpětného doplnění

Protože pipeline pro doplnění historických dat začíná od nejnovějšího offsetu v Kafce, nezachytí zprávy, které existovaly před vytvořením streamu. Chcete-li pro trénink zajistit pokrytí historickými daty, nakonfigurujte volitelný zdroj pro jejich doplnění.

Když je zdroj backfill nakonfigurovaný, Databricks spustí jednorázovou MERGE INTO úlohu, která zkopíruje řádky backfillu do tabulky pro příjem dat s record_source="backfill". Operace MERGE se spustí až poté, co kontrola překrytí potvrdí, že zdroj backfillu a forward-fill stream mají překrývající se časová razítka (viz Překrytí mezi daty backfillu a živého streamu). Pokud není splněna podmínka překrytí do 2 dnů, spustí se funkce MERGE, aby se zabránilo blokování po neomezenou dobu.

Tabulka pro backfill musí obsahovat sloupec stream_record_timestamp typu TIMESTAMP v časovém pásmu UTC. Ostatní sloupce metadat Kafka (kafka_topic, kafka_partition, kafka_offset) se předávají, pokud jsou přítomny ve zdroji pro backfill, jinak jsou nastaveny na 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"
    ),
)

Překrytí mezi zpětně doplněnými daty a daty z živého streamu

Před spuštěním příkazu MERGE mezi backfill a ingestní tabulkou kontrola překrytí porovnává časová razítka v obou tabulkách:

Maximum backfill: Maximální hodnota stream_record_timestamp ve zdroji backfill.
Minimum ingestování: Minimální stream_record_timestamp počet řádků (record_source="stream") v ingestní tabulce.

Operace MERGE proběhne, když nejnovější časové razítko backfillu přesáhne nejstarší časové razítko ingestní tabulky alespoň o 1 hodinu. Tím se zajistí, že tabulka příjmu dat neobsahuje žádné mezery. Pokud není splněna podmínka překrytí do 2 dnů, spustí se funkce MERGE, aby se zabránilo blokování po neomezenou dobu.

Protože ingestní kanál začíná od nejnovějšího offsetu v systému Kafka, zachycuje pouze zprávy, které přijdou poté, co je datový proud vytvořen. Zdroj pro zpětné doplnění dat musí obsahovat data zasahující do časového rozsahu příjmu dat — nejen do času vytvoření datového proudu.

Pokud například vytvoříte datový proud v 15:00, kanál forward-fill začne číst zprávy od 15:00 dále. Zdroj pro backfill musí obsahovat data s časovými razítky minimálně do 16:00 (1 hodinu po začátku forward-fillu), aby kontrola překrytí proběhla úspěšně. To znamená, že byste měli aktualizovat tabulku backfill po 14:00, aby se zajistilo, že tabulka příjmu dat neobsahuje žádné mezery.

Deduplication

Pomocí deduplication_columns určete cesty ke sloupcům pro identifikaci duplicitních řádků při ingesti dat mezi daty backfillu a streamovanými daty forward-fillu. Pro vnořená pole používejte tečkovou notaci (například "value.user_id").

Zvolte sloupce odstranění duplicitních dat na základě vašich dat:

Pokud každý záznam ve vašem streamu obsahuje jedinečný identifikátor (například value.transaction_id), použijte tento sloupec pro odstranění duplicitních dat.
Pokud zdroj backfill obsahuje kafka_partition a kafka_offset sloupce, použijte je k jednoznačné identifikaci každého záznamu.
Pokud nejsou zadány žádné sloupce odstranění duplicit, výchozí klíč odstranění duplicit je úplná kombinace , keyvaluea stream_record_timestamp. Toto se nedoporučuje, protože tato striktní shoda kritérií může snadno vést k duplicitám.

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

Správa datových proudů

Získání datového proudu

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

Seznam streamů

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

Nastavte include_schemas=True tak, aby zahrnovalo úplné podrobnosti schématu. Schémata můžou být velká a výsledkem může být dlouhotrvající operace. Pokud chcete načíst schémata jednotlivě, použijte get_stream.

Odstranění datového proudu

Odstraněním datového proudu se odstraní také kanál příjmu dat a tabulka příjmu dat.

Warning

Všechny modely nebo funkce, které odkazují na odstraněný stream, už nebudou mít přístup k podkladovým datům datového proudu. Před odstraněním vytvořte kopii tabulky příjmu dat, pokud tato data potřebujete, ale stream už nepotřebujete.

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

Příklad notebooku

Kompletní příklad, který vytvoří Stream, definuje funkce streamování a nasadí do obslužného koncového bodu, najdete v následujícím poznámkovém bloku:

Notebook s rychlým úvodem do Streaming Feature Views

Pořiďte si notebook

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-07-13