Konfigurowanie strumienia

Stream oznacza zewnętrzne źródło danych strumieniowych, takie jak Apache Kafka. Strumienie przechowują parametry połączenia, dane uwierzytelniające, schematy i konfigurację pozyskiwania danych. Po utworzeniu strumienia można odwoływać się do niego przy użyciu definicji widoku funkcji w celu utworzenia funkcji przesyłania strumieniowego w czasie rzeczywistym.

Strumienie mają trzyczęściowe nazwy (catalog.schema.stream_name). Dostęp do strumienia jest kontrolowany przez skojarzoną z nim tabelę pozyskiwania danych. Szczegóły można znaleźć w sekcji Spożycie i uzupełnienie .

Requirements

• Uruchamianie poleceń notesu: bezserwerowy lub klasyczny klaster obliczeniowy z uruchomionym środowiskiem Databricks Runtime 17.0 ML lub nowszym.
• Pakiet feature-engineering-client Python w wersji 0.18.0 lub wyższej musi być zainstalowany.

Łączenie ze źródłami strumieniowymi

Przed zdefiniowaniem funkcji przesyłania strumieniowego połącz się i przetestuj połączenie potoku przesyłania strumieniowego Lakeflow z brokerem platformy Kafka. Feature Store opiera się na bezserwerowym SDP, co oznacza, że będzie potrzebny mechanizm do połączenia klasycznych zasobów obliczeniowych (brokera lub punktu końcowego) z bezserwerowymi zasobami obliczeniowymi Databricks. Odbywa się to za pomocą produktów takich jak privatelink lub poprzez udostępnienie klasycznego systemu obliczeniowego z publicznego internetu.

Tworzenie strumienia

Użyj create_stream(), aby utworzyć nowy Stream. Usługa Stream wymaga czterech składników konfiguracji:

• Konfiguracja źródła: Określa platformę streamingową oraz szczegóły specyficzne dla źródła, takie jak subskrypcja tematu dla źródła Kafki.
• Konfiguracja połączenia: określa sposób nawiązywania połączenia i uwierzytelniania z platformą przesyłania strumieniowego, w tym serwerów bootstrap i poświadczeń.
• Konfiguracja schematu: definiuje strukturę kluczy i wartości komunikatów.
• Konfiguracja pozyskiwania danych: określa, gdzie i w jaki sposób pozyskiwane są dane strumieniowe. Szczegóły można znaleźć w sekcji Spożycie i uzupełnienie .

Informacje na temat source_config konfiguracji specyficznej dla źródła i konfiguracji połączenia, a także pełny przykład create_stream() znajdziesz w Apache Kafka. Opcje schematu i pozyskiwania danych są współdzielone między źródłami.

Apacz Kafka

Aby strumieniować dane z Apache Kafka, użyj KafkaStreamConfig jako konfiguracji źródła oraz połączenia Unity Catalog do uwierzytelniania. Zobacz Przesyłanie strumieniowe w bezserwerowym środowisku obliczeniowym oraz Łączenie z Apache Kafka, aby uzyskać informacje o łączeniu z platformą 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"
        ),
    ),
)

Tryby subskrypcyjne Kafka

Tryb subskrypcji określa sposób wybierania tematów platformy Kafka do korzystania z usługi Stream. Obsługiwane są trzy tryby:

Tryb	Description	Przykład
subscribe	Rozdzielona przecinkami lista nazw tematów	KafkaSubscriptionMode(subscribe="topic1,topic2")
subscribe_pattern	Nazwy tematów pasujące do wyrażenia regularnego Java	KafkaSubscriptionMode(subscribe_pattern="events-.*")
assign	Kod JSON określający przypisania partycji tematu	KafkaSubscriptionMode(assign='{"my-topic": [0, 1, 2]}')

Uwierzytelnianie w Kafka

Połączenie z usługą Unity Catalog (zalecane)

Użyj połączenia Unity Catalog, aby uwierzytelnić się w klastrze Kafka. Jest to zalecane podejście do uwierzytelniania zarządzanego. Aby utworzyć połączenie, zobacz Tworzenie połączenia. Twórca strumienia musi mieć uprawnienie USE CONNECTION do połączenia. Każdy użytkownik, który realizuje funkcje z Streamem jako źródłem, musi również mieć USE CONNECTION na połączeniu.

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

Połączenie obsługuje zarówno uwierzytelnianie IAM (service credential), jak i SASL.

IAM (poświadczenie służbowe)

Uwierzytelnij się przy użyciu poświadczeń usługi Unity Catalog, na przykład aby połączyć się z usługą Amazon MSK za pomocą mechanizmu IAM. Aby utworzyć poświadczenia usługi, zobacz Tworzenie poświadczeń usługi. Ustaw nazwę poświadczenia usługi z opcją credential :

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

Oprócz USE CONNECTION dla połączenia tożsamości korzystające z poświadczenia usługi potrzebują również ACCESS na tym poświadczeniu. Nadaj ACCESS do wskazanego poświadczenia usługi twórcy strumienia oraz każdej tożsamości, która materializuje funkcje za pomocą strumienia. Zobacz Przyznawanie uprawnień do używania poświadczenia usługi w celu uzyskania dostępu do zewnętrznej usługi w chmurze.

SASL

Uwierzytelnianie SASL wykorzystuje nazwę użytkownika i hasło. Ustaw sasl_mechanism na jedną z następujących wartości:

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

Podaj poświadczenia za pomocą opcji user i password. Połączenie bezpiecznie przechowuje te dane uwierzytelniające.

Poniższy przykład wykorzystuje SASL/SCRAM. Dla SASL/PLAIN ustaw sasl_mechanism na 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>'
)

Bezpośredni mTLS

W przypadku bezpośredniego uwierzytelniania mTLS podaj pliki keystore i truststore przechowywane w woluminie Unity Catalog, z hasłami, do których odwołują się zakresy wpisów tajnych Databricks. Aby uzyskać więcej informacji na temat uwierzytelniania SSL na platformie Kafka, zobacz Używanie protokołu SSL do nawiązywania połączenia Azure Databricks z platformą 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"
        ),
    ),
)

Konfiguracja schematu

Zdefiniuj strukturę kluczy i wartości wiadomości, aby definicje odbiorców i cech mogły odczytywać poszczególne pola. W przypadku źródeł payload_schema platformy Kafka odpowiada wartości komunikatu platformy Kafka ( value w modelu klucz-wartość platformy Kafka) i key_schema odpowiada kluczowi komunikatu platformy Kafka. Należy podać co najmniej jedno z pól payload_schema lub key_schema.

Każdy SchemaConfig z nich akceptuje jeden z trzech formatów, odpowiadających sposobowi serializacji wiadomości json_schemaprzez źródło: , avro_schema, lub proto_schema. Jeśli dla klucza lub ładunku nie podano żadnego schematu, jest on traktowany jako prosty ciąg.

Przykłady kodu w tej sekcji używają schematów zadeklarowanych bezpośrednio przy użyciu DirectSchemas, w których schemat jest podany jako ciąg znaków. Aby zarządzać schematami za pomocą zewnętrznego rejestru schematów, szczegóły można znaleźć w rejestrze schematów .

Schemat systemu JSON

Dostarcz ciąg schematu JSON do 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"}'
    ),
)

Schemat Avro

Dostarcz ciąg schematu Avro do avro_schema. Obsługiwane są typy logiczne Avro, w tym timestamp-millis, date, oraz 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"}}'
            '  ]'
            '}'
        )
    ),
)

Schemat Protobuf

Przekaż do ProtoSchemaSpecproto_schema tekst źródłowy Protocol Buffers.proto oraz nazwę komunikatu payload. Zaimportuj ProtoSchemaSpec z databricks.feature_engineering.entitiespliku .

message_name musi być w pełni kwalifikowaną nazwą komunikatu, wraz z package zadeklarowaną w tekście .proto (na przykład com.example.Event, a nie Event). Obsługiwane są zarówno składnie proto2, jak i proto3.

Obsługiwane są google.protobuf.Timestamp oraz typy opakowujące dla typów skalarnych (StringValue, Int32Value itd.), a ich importy są rozwiązywane automatycznie. Inne znane typy, takie jak Duration, Struct, i Any, są odrzucane; koduj te wartości jako wspierany skalar lub komunikat. Typy skalarne fixed32 i fixed64 oraz map z kluczami innymi niż ciągi znaków również nie są obsługiwane.

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

Dekodowanie danych za pomocą schematów

Databricks dekoduje każdą wiadomość za pomocą funkcji Spark from_avro, from_protobuf i from_json. Następujące zachowania mają zastosowanie niezależnie od tego, czy deklarujesz schemat w linii, czy rozwiązujesz go z rejestru schematów:

• Zdeformowane rekordy. Dekodowanie używa trybu PERMISSIVE, więc rekord, który nie jest zgodny ze swoim schematem, jest dekodowany do wartości null zamiast powodować błąd strumienia.
• Związki zawodowe Avro. Unia kilku typów rekordów jest dekodowana do struktury z jednym polem dla każdego typu rekordu, z których każde nosi nazwę odpowiadającą rekordowi Avro.
• Typy protobufów. Liczby całkowite bez znaku są dekodowane do szerszego typu ze znakiem (na przykład uint32 do BIGINT oraz uint64 do DECIMAL(20,0)), pola enum są dekodowane do ich nazw tekstowych, a skalarne typy opakowujące (na przykład StringValue i Int32Value) są dekodowane do kolumny dopuszczającej wartość null typu opakowanego.

Rejestr schematów

Rejestry schematów przechowują i wersjonują schematy używane przez producentów i konsumentów danych strumieniowych, wymuszając reguły zgodności w miarę ich ewolucji. Gdy zewnętrzny rejestr schematu zostanie skonfigurowany, Feature Store odczytuje schemat z rejestru i używa go do dekodowania wiadomości streamingowej. Nie deklarujesz schematu w linii w Streamie, korzystając z rejestru schematów.

Wsparcie rejestru schematów ma następujące ograniczenia:

• Obsługiwane tylko dla streamów Kafki.
• Obsługiwany jest tylko Confluent Schema Registry
• Obsługiwane są tylko formaty Avro i Protobuf . Aby czytać komunikaty JSON, zadeklaruj schemat w linii. Zobacz schemat JSON.
• Każdy strumień jest połączony dokładnie z jednym subjectem Confluent dla wartości komunikatu oraz jednym dla klucza komunikatu (jeśli został podany). Tematy strumienia zawierające wiele rekordów schematu nie stanowią obsługiwanej konfiguracji. Jeśli Twój Strumień łączy się z tematami zawierającymi wiele schematów, rekordy niezgodne ze schematem dla danego tematu są dekodowane jako null.

Połącz się z rejestrem schematów

Udostępnij szczegóły połączenia rejestru jako opcje na połączeniu Kafka Unity Catalog i przechowuj sekret API rejestru w skali tajnego Databricks. Tożsamość uruchomieniowa strumienia musi mieć READ uprawnienie do zakresu wpisów tajnych, ponieważ potok pozyskiwania danych odczytuje wpis tajny w czasie wykonywania. Aby dowiedzieć się, jak utworzyć i skonfigurować połączenie, zobacz: Utwórz połączenie.

Dodaj schema_registry_url, schema_registry_api_key, i schema_registry_api_secret opcje do połączenia używanego do uwierzytelniania. Poniższy przykład tworzy połączenie Kafki, które uwierzytelnia się do brokera za pomocą uwierzytelniania usługi Unity Catalog oraz do rejestru za pomocą klucza 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>')
)

Ustaw schema_registry_api_secret zarówno opcję połączenia Kafka, jak i referencję do zakresu tajnego w strumieniu na ten sam sekret.

Stwórz strumień korzystający z rejestru schematu

Przekaż SchemaRegistryConfig jako schema_config. Odwołaj się do sekretu interfejsu API rejestru za pomocą api_secret_ref, a temat i format określ za pomocą key_schema_locator dla wartości wiadomości lub payload_schema_locator dla klucza wiadomości. Należy podać co najmniej jeden lokalizator.

Zwróć uwagę na różnice w porównaniu z przykładami schematów bezpośrednich w sekcji konfiguracji schematów . Korzystając z rejestru schematów, nie udostępniasz schematu w linii na strumieniu do schema_config. Zamiast tego podajesz SchemaRegistryConfig, który identyfikuje schemat w rejestrze.

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

Podmiot Confluent to nazwany zakres, w którym rejestrowana jest historia wersji schematu i wymuszona jest zgodność. Ustaw subject na nazwę odpowiedniego zakresu, która jest zwykle określana na podstawie strategii nazewnictwa tematu:

• TopicNameStrategy (domyślnie, wyprowadza podmiot z nazwy tematu): <topic>-value dla wartości i <topic>-key dla klucza. Na przykład schemat wartości dla tematu transactions wykorzystuje podmiot transactions-value.
• RecordNameStrategy (wyprowadza nazwę subjectu z nazwy rekordu schematu, niezależnie od topicu): w pełni kwalifikowana nazwa rekordu, na przykład com.example.Payment. Jest to przestrzeń nazw i nazwa rekordu w Avro lub pakiet i nazwa wiadomości w Protobuf.
• TopicRecordNameStrategy (łączy nazwy tematów i rekordów): <topic>-<fully-qualified-record-name>, takie jak transactions-com.example.Payment.

Ciąg format jest wymagany. Ustaw tę wartość na SchemaLocatorFormat.FORMAT_AVRO lub SchemaLocatorFormat.FORMAT_PROTOBUF, aby odpowiadała sposobowi serializacji tematu.

Ewolucja schematu

Potok przetwarzania danych wejściowych ustala bieżący schemat obiektu w momencie uruchomienia. Gdy rejestrujesz nową, kompatybilną wsteczną wersję schematu na danym obszarze w rejestrze schematów, działający potok nadal korzysta z wersji, od której zaczął.

W przypadku strumieni wspieranych przez rejestr schematów potok pozyskiwania danych uruchamia się ponownie automatycznie co kilka godzin. Przy każdym ponownym uruchomieniu pobiera najnowszą wersję schematu tematu, a w tabeli pozyskiwania danych pojawiają się nowe lub zmienione pola.

Strumienie, które używają schematów bezpośrednio w strumieniu zamiast korzystać z rejestru schematów, ewoluują schemat za pomocą update_stream. Zobacz Aktualizowanie strumienia.

Aby dowiedzieć się, jak potok obsługuje rekordy, które nie odpowiadają aktualnie używanemu schematowi, zobacz Dekodowanie danych za pomocą schematów.

Filtrowanie rekordów według typu

Strumień dekoduje każdy rekord za pomocą pojedynczego schematu klucza i wartości (jeśli jest dostępny), niezależnie od tego, czy podajesz je bezpośrednio, czy używasz rejestru schematów. Ponieważ temat może zawierać więcej niż jeden typ rekordu, a strumienie mogą subskrybować wiele tematów, użyj do record_type_filter wyboru, które rekordy z tematu należą do tego strumienia.

Zapewnij wyrażenie SQL, które odnosi się do zdekodowanych pól z notacją kropkową, na przykład value.event_type = 'transaction'. Zapisy, które nie pasują do filtra, są ignorowane. Nie są zapisywane w tabeli pobierania i nie są używane w materializacji. Aby utworzyć strumień dla innych typów rekordów, stwórz osobny strumień z innym 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'",
)

Nawet bez record_type_filter, dekodowanie nigdy nie powoduje awarii strumienia. Rekord, który nie odpowiada skonfigurowanemu schematowi, jest dekodowany w sposób tolerancyjny. Rekordy są dekodowane na jeden z następujących sposobów:

• Do wiersza zawierającego wartości NULL dla pól wymaganych przez schemat, których brakuje w rekordzie (JSON, Avro i Protobuf).
• Do wiersza z wartościami należącymi do innego typu rekordu (tylko Avro i Protobuf).

Aby określić, czy wiersz w tabeli pozyskiwania danych należy do oczekiwanego typu rekordu, użyj jednej z poniższych metod sprawdzania:

• Sprawdź, czy pole jest równe wartości oczekiwanej, na przykład value.event_type = 'transaction' (preferowane dla Avro i Protobuf).
• Sprawdź, czy pole nie jest value.activity_id IS NOT NULL, na przykład NULL.

Zaleca się używanie record_type_filter z oddzielnymi strumieniami, gdy schematy znacznie różnią się między typami rekordów w temacie lub gdy chcesz niezależnie zarządzać dostępem do każdego typu rekordu. Aby ograniczyć koszty, Databricks zaleca utrzymywanie małej liczby elementów Streams, ponieważ każdy strumień ma osobny potok pozyskiwania danych i tabelę pozyskiwania danych. Każdy strumień również wykorzystuje oddzielne zasoby obliczeniowe podczas materializacji. Do materializacji można używać filtrów specyficznych dla funkcji.

record_type_filter różni się od funkcji filter_condition. record_type_filter jest ustawiona na strumieniu i kontroluje, które rekordy są pobierane i dostępne dla wszystkich cech korzystających ze strumienia jako źródła, natomiast filter_condition jest ustawiona na pojedynczej funkcji i filtruje wiersze przed agregacją. Zobacz Warunki filtrów na źródłach streamingowych, aby uzyskać więcej szczegółów na .filter_condition

Pozyskiwanie i uzupełnianie danych

Parametr ingestion_config służy do konfigurowania sposobu przechwytywania i przechowywania danych strumienia na potrzeby trenowania i obsługi.

Dostęp do strumienia jest regulowany przez tabelę pozyskiwania danych:

• SELECT w tabeli pozyskiwania danych przyznaje uprawnienia do odczytu strumienia Stream.
• MANAGE w tabeli pozyskiwania udziela dostępu do usuwania.

Aby uzyskać więcej informacji na temat uprawnień tabel, zobacz Tabela i Informacje o uprawnieniach w Unity Catalog.

Potok pozyskiwania danych

Po utworzeniu strumienia Databricks uruchamia zarządzany potok pobierania, który nieprzerwanie odczytuje wiadomości ze strumienia źródłowego i zapisuje je do tabeli Delta (tabeli pobierania). Potok przetwarzania zaczyna działanie od najnowszej pozycji źródła i działa nieprzerwanie, rejestrując tylko nowe wiadomości, które pojawiają się po utworzeniu strumienia. Ta tabela pozyskiwania służy do trenowania z funkcjami przesyłania strumieniowego. Po usunięciu strumienia jego potok przetwarzania i tabela przetwarzania również zostaną usunięte.

Miejsce docelowe importu

ingestion_destination określa trzyczęściową nazwę tabeli Delta, do której są zapisywane dane strumieniowe.

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

Schemat tabeli pozyskiwania danych

Tabela pobierania zawiera dane wiadomości wraz z kolumnami metadanych. Wspólne kolumny są obecne dla każdego źródła; kolumny kafka_* są obecne tylko dla strumienia Kafka.

Column	Typ	Source	Description
key	Różni się (od key_schema)	Common	Klucz wiadomości, sstrukturyzowany zgodnie ze schematem, który podałeś.
value	Różni się (od payload_schema)	Common	Wartość wiadomości (payload), sstrukturyzowana zgodnie ze schematem, który podałeś.
stream_record_timestamp	TIMESTAMP	Common	Sygnatura czasowa rekordu. Dla danych uzupełniających do przodu jest to znacznik czasu pobierania źródła. W przypadku danych uzupełniających są one dostarczane przez klienta.
record_source	STRING	Common	Albo "stream" (uzupełnianie do przodu na podstawie strumienia na żywo), albo "backfill" (ze źródła danych uzupełniających wstecz).
kafka_topic	STRING	Kafka	Temat Kafka, z którego odczytano rekord.
kafka_partition	INT	Kafka	Partycja Kafka, z której rekord został odczytany.
kafka_offset	LONG	Kafka	Przesunięcie platformy Kafka rekordu w ramach partycji.

Źródło wypełniania

Ponieważ potok uzupełniania do przodu zaczyna się od najnowszej pozycji w źródle, nie przechwytuje komunikatów, które istniały przed utworzeniem strumienia. Aby zapewnić dostęp do danych historycznych na potrzeby szkolenia, skonfiguruj opcjonalne źródło uzupełniania danych historycznych.

Po skonfigurowaniu źródła uzupełniania danych usługa Databricks uruchamia jednorazowe zadanie MERGE INTO, które kopiuje wiersze uzupełnianych danych do tabeli ingestii za pomocą polecenia record_source="backfill". Funkcja MERGE jest uruchamiana dopiero po upewnieniu się, że źródło wypełniania wstecznego i strumień wypełniania dalej mają nakładające się znaczniki czasu (zobacz Nakładanie się między danymi wypełniania i strumienia na żywo). Jeśli warunek pokrywania się nie zostanie spełniony w ciągu 2 dni, operacja MERGE i tak zostanie uruchomiona, aby uniknąć blokowania w nieskończoność.

Tabela wypełniania musi zawierać kolumnę stream_record_timestamp typu TIMESTAMP w strefie czasowej UTC. Inne kolumny metadanych są przenoszone, jeśli występują w źródle backfill, a w przeciwnym razie mają wartość NULL. Dla Kafki są to kafka_topic, kafka_partition, oraz 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"
    ),
)

Nakładanie się między danymi wypełniania i transmisji strumieniowej na żywo

Przed wykonaniem operacji MERGE między uzupełnianiem danych historycznych a tabelą ładowania danych sprawdzanie nakładania się porównuje znaczniki czasu w obu tabelach:

• Maksimum uzupełniania wstecznego: Maksymalna wartość stream_record_timestamp w źródle uzupełniania wstecznego.
• Minimum ładowania: Minimalna stream_record_timestamp liczba wierszy (record_source="stream") w tabeli ładowania.

Operacja MERGE jest wykonywana, gdy najnowszy znacznik czasu backfillu jest późniejszy od najwcześniejszego znacznika czasu tabeli ingestii o co najmniej 1 godzinę. To nakładanie się zapewnia, że w tabeli importu nie ma żadnych luk. Jeśli warunek pokrywania się nie zostanie spełniony w ciągu 2 dni, operacja MERGE i tak zostanie uruchomiona, aby uniknąć blokowania w nieskończoność.

Ponieważ potok pobierania zaczyna się od najnowszej pozycji w źródle, rejestruje tylko wiadomości przychodzące po utworzeniu strumienia. Źródło danych do uzupełnienia historycznego musi zawierać dane, które obejmują zakres czasu pozyskiwania danych — a nie tylko sięgają do momentu utworzenia strumienia.

Jeśli na przykład tworzysz strumień o godzinie 15:00, potok wypełniania dalej rozpocznie odczytywanie komunikatów od 15:00. Źródło wypełniania musi zawierać dane ze znacznikami czasu do co najmniej 16:00 (1 godzina po rozpoczęciu wypełniania do przodu), aby spełnić nakładające się sprawdzanie. Oznacza to, że należy zaktualizować swoją tabelę uzupełnień po godzinie 16:00, aby upewnić się, że w tabeli ładowania nie ma luk.

Deduplication

Użyj deduplication_columns, aby określić ścieżki kolumn służące do identyfikowania zduplikowanych wierszy podczas pozyskiwania danych między danymi strumieniowymi backfill i forward-fill. Użyj notacji kropkowej dla zagnieżdżonych pól (na przykład "value.user_id").

Wybierz kolumny do usuwania duplikatów w oparciu o dane:

• Jeśli każdy rekord w strumieniu zawiera unikatowy identyfikator (na przykład value.transaction_id), użyj tej kolumny do deduplikacji.
• Jeśli źródło danych do uzupełnienia zawiera kolumny kafka_partition i kafka_offset, użyj ich do jednoznacznego identyfikowania każdego rekordu.
• Jeśli nie określono żadnych kolumn deduplikacji, domyślny klucz deduplikacji jest pełną kombinacją , keyvaluei stream_record_timestamp. Nie jest to zalecane, ponieważ te ścisłe kryteria dopasowania mogą łatwo prowadzić do duplikatów.

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

Przypisywanie kosztów

Ustaw tags i budget_policy_id w elemencie IngestionConfig, aby przypisać koszt zarządzanego pozyskiwania danych strumienia. Azure Databricks stosuje je do potoku pozyskiwania danych Lakeflow oraz jego zadań forward-fill i backfill, gdy tworzony jest obiekt Stream.

Przykład limitów tagów oraz informacji o tym, jak sprawdzić przypisane koszty, znajdziesz w artykule Przypisywanie kosztów za pomocą tagów i zasad użycia usług bezserwerowych.

Wyklucz kolumny ze strumienia

Używam do excluded_columns usuwania konkretnych kolumn ze strumienia, których nie chcesz pobierać. Wykluczona kolumna nie jest zapisywana do tabeli ingestii i nie może być używana przez cechę ani na potrzeby trenowania.

Określ każdą kolumnę przy użyciu notacji kropkowej w kluczu lub wartości komunikatu, na przykład value.user.email lub key.account_id. Kolumny te są usuwane z dekodowanego key i value pozyskiwania powrotem, zasypywaniem i materializacją. Jeśli ścieżka wskazuje na strukturę, wszystkie jej zagnieżdżone pola również są usuwane (na przykład usunięcie value.address powoduje również usunięcie value.address.city i 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"],
)

W przypadku użycia schematów bezpośrednich wykluczona kolumna musi już istnieć w schemacie klucza lub wartości, w przeciwnym razie działanie create_stream zakończy się niepowodzeniem. Korzystając z rejestru schematów, możesz wykluczyć kolumnę, zanim zostanie utworzona. Kolumna wykluczona również nie może być kolumną deduplikacji, ponieważ kolumny deduplikacyjne są wymagane do identyfikacji zduplikowanych wierszy. Każda cecha, która odnosi się do kolumny wyłączonej (na przykład jako jednostka, szereg czasowy lub wejście), nie zostaje utworzona.

Możesz zmienić kolumny wykluczone strumienia po utworzeniu za pomocą update_stream, zarówno na strumieniach podtrzymywanych przez schemat bezpośredni, jak i rejestrowany przez schemat. Zobacz Aktualizowanie strumienia, aby uzyskać więcej szczegółów.

Zarządzaj strumieniami

Pobierz strumień

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

Wyświetlanie listy strumieni

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

Ustaw wartość include_schemas=True , aby uwzględnić pełne szczegóły schematu. Schematy mogą być duże i może to spowodować długotrwałą operację. Aby pobrać schematy indywidualnie, użyj polecenia get_stream.

Zaktualizuj stream

Użyj update_stream do zmiany strumienia po utworzeniu. Przekaż schema_config, aby rozwijać schemat bezpośredni, excluded_columns, aby zmienić, które kolumny są pomijane, lub oba. Aktualizacja innych pól nie jest obsługiwana. Zamiast tego stwórz nowy Stream.

Aktualizacja strumienia uruchamia jego potok pobierania, więc zmiana wchodzi w życie. Spożycie zwykle wraca w ciągu kilku minut.

Ewolucja schematu bezpośredniego

Dla strumienia używającego schematów bezpośrednich przekażmy a DirectSchemas do schema_config. Ustaw payload_schema, key_schema, lub oba te elementy. Strona, której nie ustawisz, pozostaje niezmieniona. Strumienie wspierane przez rejestr schematów odrzucają aktualizację schema_config i należy je zamiast tego zmieniać za pośrednictwem rejestru.

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

Aktualizacje schematu muszą być kompatybilne wstecz, aby działający potok przetwarzania danych mógł dalej dekodować istniejące rekordy i zapisywać je do tabeli przetwarzania danych. Wszelkie inne zmiany są odrzucane.

To, co jest dozwolone, zależy od formatu:

• JSON i Protobuf: dodaj pola opcjonalne, usuń pola i poszerz typ pola (na przykład z int do bigint). Protobuf pozwala także na zmianę kolejności pól.
• Avro: pozwala jedynie na poszerzanie intlong i usunięcie pola końcowego, którego bajty nie odczytują późniejszego pola. Aby swobodniej rozwijać schemat Avro, użyj strumienia wspieranego przez rejestr schematów.

Dodawanie pól zwiększa zdekodowane key i value strukturalne struktury tabeli ingestów. Wiersze zapisane przed aktualizacją zachowują swoją pierwotną strukturę, a dodane pola mają wartość NULL w tych wcześniejszych wierszach. Usuwanie i zmiany typów obowiązują tylko dla rekordów wnieśliwych po aktualizacji.

Zmień wykluczone kolumny

Przekaż pełny nowy zestaw ścieżek kolumn do excluded_columns, który zastępuje istniejący zbiór. Przekaż pustą listę ([]), aby usunąć wszystkie wykluczenia. Szczegóły dotyczące tego zachowania można znaleźć w artykule Wyklucz kolumny ze strumienia.

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

Zmiana kolumn wykluczonych jest dostępna tylko w kierunku przekazywania. Nowo wykluczone kolumny przestają być zapisywane (pojawiając się jako NULL), a nowo uwzględnione kolumny od tego momentu zaczynają być uzupełniane, podczas gdy wcześniej zapisane wiersze pozostają bez zmian. Aby zapobiec wchłonięciu nowej kolumny:

• Rejestr schematów: najpierw dodaj kolumnę do excluded_columns i poczekaj na ponowne uruchomienie potoku przetwarzania danych, a następnie zarejestruj w rejestrze nową wersję schematu.
• Schematy bezpośrednie: dodawaj kolumnę do schema_config i do excluded_columns tego samego update_stream połączenia.

Usuń strumień

Usunięcie strumienia usuwa również potok pozyskiwania danych i tabelę pozyskiwania danych.

Warning

Wszystkie modele lub funkcje odwołujące się do usuniętego strumienia nie będą już miały dostępu do danych źródłowych strumienia. Utwórz kopię tabeli pozyskiwania danych przed jej usunięciem, jeśli potrzebujesz tych danych, ale nie potrzebujesz już strumienia.

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

Przykładowy notatnik

Pełny przykład tworzenia obiektu Stream, definiowania funkcji strumieniowych i wdrażania w punkcie końcowym serwującym znajduje się w następującym notatniku:

Notatnik szybkiego startu dotyczący widoków cech strumieniowych

Pobierz laptopa