ストリームを設定する

備考

プレビュー

この機能は パブリック プレビュー段階です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

ストリームは、Apache Kafkaなどの外部ストリーミングデータソースを表します。ストリームは、接続の詳細、認証、スキーマ、および取り込み設定を保存します。ストリームが作成された後、特徴量ビューの定義を使用して参照し、リアルタイムストリーミング特徴量を作成できます。

ストリームには3部分からなる名前（catalog.schema.stream_name）があります。ストリームへのアクセスは、関連付けられた取り込みテーブルによって制御されます。詳細については、取り込みとバックフィルを参照してください。

要件

• ノートブックコマンドの実行には：サーバレス、またはDatabricks Runtime 17.0 機械学習以上を実行しているクラシックなコンピュートクラスターが必要です。
• feature-engineering-client Python パッケージのバージョン 0.17.0 以降がインストールされている必要があります。

ストリームソースへの接続

ストリーミング特徴量を定義する前に、ストリーミング LakeFlow Pipelines 接続を Kafka ブローカーに接続してテストします。Feature Store はServerless SDP に依存しているため、クラシックコンピュート (ブローカーまたはEndpoint) を Databricks Serverless コンピュートに接続する仕組みが必要になります。これは、PrivateLink などの製品を使用するか、クラシックコンピュートをパブリックインターネットからアクセス可能にすることで行われます。

ストリームを作成

新しいストリームを作成するには、create_stream()を使用します。ストリームには4つの構成コンポーネントが必要です：

• ソース設定 ：ストリーミングプラットフォームと、Kafka ソースのトピックサブスクリプションなどのソース固有の詳細を指定します。
• 接続構成 ：ブートストラップサーバと資格情報を含め、ストリーミングプラットフォームに接続および認証する方法を指定します。
• スキーマ設定：メッセージのキーと値の構造を定義します。
• Ingestion config : ストリームデータの取り込み場所と方法を指定します。詳細については、「取り込みとバックフィル」を参照してください。

ソース固有の source_config と接続のセットアップ、および完全な create_stream() の例については、Apache Kafka を参照してください。スキーマおよび取り込みオプションは、ソース間で共有されます。

Apache Kafka

Apache Kafka からストリーミングするには、ソース設定として KafkaStreamConfig を使用し、認証に Unity Catalog 接続を使用します。Kafka の接続については、ServerlessコンピュートでのストリーミングおよびApache Kafka への接続を参照してください。

Python

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 サブスクリプション モード

サブスクリプションモードは、ストリームが消費するKafkaトピックをどのように選択するかを指定します。3つのモードがサポートされています：

モード	説明	例
subscribe	トピック名のコンマ区切りリスト	KafkaSubscriptionMode(subscribe="topic1,topic2")
subscribe_pattern	Java 正規表現パターンによるトピック名の一致	KafkaSubscriptionMode(subscribe_pattern="events-.*")
assign	トピックパーティションの割り当てを指定する JSON	KafkaSubscriptionMode(assign='{"my-topic": [0, 1, 2]}')

モード	説明	例
subscribe	トピック名のコンマ区切りリスト	KafkaSubscriptionMode(subscribe="topic1,topic2")
subscribe_pattern	Java 正規表現パターンによるトピック名の一致	KafkaSubscriptionMode(subscribe_pattern="events-.*")
assign	トピックパーティションの割り当てを指定する JSON	KafkaSubscriptionMode(assign='{"my-topic": [0, 1, 2]}')

Kafka 認証

Unity Catalog接続 (推奨)

Unity Catalog接続を使用して、Kafkaクラスターへの認証を行います。これは、管理された認証において推奨されるアプローチです。接続を作成するには、接続の作成を参照してください。ストリームの作成者は、接続に対する USE CONNECTION 権限を持っている必要があります。ストリームをソースとしてフィーチャをマテリアライズするユーザーは、接続に対する USE CONNECTION 権限も持っている必要があります。

Python

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

この接続は、IAM（サービス資格情報）認証とSASL認証の両方をサポートしています。

IAM（サービス資格情報）

Unity Catalog サービス資格情報を使用して認証します（例：IAM を使用して Amazon MSK に接続する場合）。サービス資格情報を作成するには、サービス資格情報の作成を参照してください。credential オプションを使用して、サービス資格情報名を設定します：

SQL

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

接続に対するUSE CONNECTIONに加えて、サービス資格情報を使用するIDには、その資格情報に対するACCESSが必要です。参照されるサービス資格情報に対するACCESSを、ストリームの作成者およびストリームで特徴量をマテリアライズするすべてのIDに付与します。サービス資格情報を使用して外部クラウドサービスにアクセスするためのアクセス許可を付与するを参照してください。

SASL

SASL認証では、ユーザー名とパスワードを使用します。sasl_mechanismには、次のいずれかを設定してください：

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

user および password オプションを使用して資格情報を提供します。接続は、これらの資格情報を安全に保存します。

次の例では、SASL/SCRAMを使用しています。SASL/PLAINの場合は、sasl_mechanismをPLAINに設定します。

SQL

CREATE CONNECTION IF NOT EXISTS `my-kafka-connection`
TYPE KAFKA
OPTIONS (
    bootstrap_servers '<bootstrap_servers>',
    sasl_mechanism 'SCRAM-SHA-512',
    user '<username>',
    password '<password>'
)

直接mTLS

直接mTLS認証の場合、Unity Catalogボリュームに保存されているキーストアファイルとトラストストアファイルを提供し、パスワードはDatabricksシークレットスコープを通じて参照されます。KafkaとのSSL認証の詳細については、SSLを使用してDatabricksをKafkaに接続するを参照してください。

Python

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

スキーマ構成

取り込みおよび特徴量の定義が個々のフィールドを読み取れるように、メッセージキーと値の構造を定義します。Kafka ソースの場合、payload_schema は Kafka メッセージ値（Kafka のキー・値モデルにおける value）に対応し、key_schema は Kafka メッセージキーに対応します。payload_schema または key_schema のいずれかを少なくとも1つ指定する必要があります。

各 SchemaConfig は、ソースがメッセージをシリアル化する方法に合わせて、json_schema、avro_schema、または proto_schema の3つの形式のいずれかを受け入れます。キーまたはペイロードのスキーマが提供されない場合、単純な文字列として扱われます。

このセクションのコード例では、DirectSchemasを使用してインラインで宣言されたスキーマを使用しており、スキーマは文字列として提供されます。外部スキーマレジストリを使用してスキーマを管理する方法については、スキーマレジストリの詳細を参照してください。

JSONスキーマ

JSONスキーマ文字列を json_schema に提供します。

Python

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

avro_schema に Avro schema 文字列を提供します。timestamp-millis、date、および decimal を含む、Avro 論理型がサポートされています。

Python

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スキーマ

ProtoSchemaSpecからproto_schemaへ、Protocol Buffers .protoのソーステキストとペイロードメッセージ名を指定します。databricks.feature_engineering.entitiesからProtoSchemaSpecをインポートします。

message_name .proto テキストで宣言された package を含む、完全修飾メッセージ名である必要があります（例：Event ではなく com.example.Event）。proto2 と proto3 の両方の構文がサポートされています。

google.protobuf.Timestamp また、スカラーラッパー型 (StringValue、 Int32Value など) もサポートされており、それらのインポートは自動的に解決されます。Duration、 Struct、 Any などのその他の既知の型は拒否されます。代わりに、それらの値をサポートされているスカラーまたはメッセージとしてエンコードしてください。fixed32 および fixed64 スカラー型と、文字列以外のキーを持つ map もサポートされていません。

Python

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

スキーマを使用したデータのデコード

Databricks は、Spark の from_json、from_avro、および from_protobuf 関数を使用して各メッセージをデコードします。スキーマをインラインで宣言する場合でも、スキーマレジストリから解決する場合でも、以下の動作が適用されます。

• 不正な形式のレコード。 デコードには PERMISSIVE モードが使用されるため、スキーマと一致しないレコードはストリームを失敗させるのではなく、null 値にデコードされます。
• Avro 共用体。 複数のレコードタイプの共用体（union）は、レコードタイプごとに1つのフィールドを持つ構造体にデコードされ、各フィールドにはそのAvroレコードの名前が付けられます。
• Protobuf 型。 符号なし整数はより広い符号付き型にデコードされ (たとえば uint32 は BIGINT に、uint64 は DECIMAL(20,0) に)、enum フィールドは文字列名にデコードされ、スカラーラッパー型 (たとえば StringValue および Int32Value) はラップされた型の NULL 許容列にデコードされます。

スキーマレジストリ

スキーマレジストリは、ストリーミングのプロデューサーとコンシューマーが使用するスキーマを保存およびバージョン管理し、スキーマの進化に合わせて互換性ルールを強制します。外部スキーマレジストリが構成されている場合、Feature Storeはレジストリからスキーマを読み取り、それを使用してストリーミングメッセージをデコードします。スキーマレジストリを使用する場合、ストリーム上でスキーマをインラインで宣言する必要はありません。

スキーマレジストリのサポートには、次の制限があります。

• Kafka ストリームでのみサポートされます。
• Confluent Schema Registry のみがサポートされています
• Avro および Protobuf 形式のみがサポートされています。JSONメッセージを読み取るには、代わりにスキーマをインラインで宣言します。JSONスキーマを参照してください。
• 各ストリームは、メッセージ値に対して1つの Confluentサブジェクトに、また（提供されている場合は）メッセージキーに対して1つのサブジェクトに接続されます。複数のスキーマレコードを含むストリームトピックは、サポートされている構成ではありません。Streamが複数のスキーマを含むトピックに接続する場合、指定されたサブジェクトのスキーマと一致しないレコードは null としてデコードされます。

スキーマレジストリへの接続

レジストリ接続の詳細をKafka Unity Catalog接続のオプションとして提供し、レジストリAPIシークレットをDatabricksのSecret Scopeに保存します。インジェクションパイプラインが実行時にシークレットを読み取るため、ストリームの実行IDにはSecret Scopeに対するREAD権限が必要です。接続の作成および設定方法については、接続の作成を参照してください。

schema_registry_url、schema_registry_api_key、および schema_registry_api_secret オプションを認証に使用される接続に追加します。次の例では、Unity Catalogのサービス資格情報を使用してブローカーに対して、またAPIキーを使用してレジストリに対して認証を行うKafka接続を作成します：

SQL

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

Kafka接続の schema_registry_api_secret オプションと、ストリーム上の Secret Scope 参照の両方を同じシークレットに設定します。

スキーマレジストリを使用するストリームを作成する

SchemaRegistryConfig を schema_config として渡します。api_secret_ref でレジストリ API シークレットを参照し、メッセージ値の場合は payload_schema_locator、メッセージキーの場合は key_schema_locator でサブジェクトと形式を識別します。少なくとも1つのロケーターを指定する必要があります。

スキーマ構成セクションの直接的なスキーマ例と比較して、ここでの違いに注意してください。スキーマレジストリを使用する場合、ストリーム上でschema_configにスキーマをインラインで提供する必要はありません。代わりに、レジストリ内のスキーマを識別するSchemaRegistryConfigを指定します。

Python

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

Confluent サブジェクトとは、スキーマのバージョン履歴が登録され、互換性が強制される名前付きスコープのことです。subject に関連するスコープの名前を設定します。これは通常、サブジェクト名の戦略から決定されます：

• TopicNameStrategy (default、トピック名からサブジェクトを派生): 値には <topic>-value、キーには <topic>-key を使用します。たとえば、トピック transactions の値スキーマはサブジェクト transactions-value を使用します。
• RecordNameStrategy （トピックとは無関係に、スキーマのレコード名からサブジェクトを導出します）：com.example.Payment のような完全修飾レコード名。これはAvroの場合はレコードの名前空間と名前、Protobufの場合はメッセージのパッケージと名前です。
• TopicRecordNameStrategy (トピック名とレコード名を組み合わせます): <topic>-<fully-qualified-record-name> (例: transactions-com.example.Payment)。

format が必要です。トピックのシリアル化方法に合わせて、SchemaLocatorFormat.FORMAT_AVRO または SchemaLocatorFormat.FORMAT_PROTOBUF に設定します。

スキーマ進化

取り込みパイプラインは、起動時にサブジェクトの現在のスキーマを解決します。スキーマレジストリのサブジェクトに新しい後方互換性のあるスキーマバージョンを登録しても、実行中のパイプラインは起動時に使用していたバージョンを引き続き使用します。

Because Databricks manages the ingestion pipeline as a serverless Lakeflow pipeline, the pipeline restarts periodically. Databricksは取り込みパイプラインをサーバーレスのLakeflow pipelineとして管理しているため、パイプラインは定期的に再起動します。次回の再起動時に、新しいスキーマバージョンが反映されます。新規または変更されたフィールドが取り込みテーブルに表示されるまで、最大1週間かかる場合があります。

パイプラインが現在使用しているスキーマと一致しないレコードをどのように処理するかについては、スキーマを使用したデータのデコードを参照してください。

取り込みとバックフィル

ingestion_configパラメーターは、ストリームデータがトレーニングおよびサービング用にキャプチャおよび保存される方法を構成します。

ストリームへのアクセスは、インジェスチョンテーブルによって管理されます：

• SELECT 取り込みテーブルに対する読み取りアクセス権をストリームに付与します。
• MANAGE 取り込みテーブルに対する削除アクセスを付与します。

テーブル情報の詳細については、テーブルおよびUnity Catalog権限リファレンスを参照してください。

取り込みパイプライン

ストリームが作成されると、Databricks は管理された取り込みパイプラインを起動します。このパイプラインは、ソースストリームからメッセージを継続的に読み取り、Delta テーブル（取り込みテーブル）に書き込みます。パイプラインはソース内の最新の位置から起動し、継続的にランを実行して、ストリームの作成後に到着した新しいメッセージのみをキャプチャします。この取り込みテーブルは、ストリーミング機能を使用したトレーニングに使用されます。ストリームが削除されると、その取り込みパイプラインと取り込みテーブルも削除されます。

取り込み先

ingestion_destinationは、ストリームデータが書き込まれる3部構成のDeltaテーブル名を指定します。

Python

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

取り込みテーブルスキーマ

取り込みテーブルには、メッセージデータとメタデータ列が含まれています。共通の列はすべてのソースに存在しますが、kafka_* の列はKafkaストリームにのみ存在します。

列	Type	ソース	説明
key	key_schemaによって異なります	ソース	指定したスキーマに従って構造化されたメッセージキー。
value	payload_schemaによって異なります	ソース	指定したスキーマに従って構造化されたメッセージ値（ペイロード）。
stream_record_timestamp	TIMESTAMP	ソース	レコードの Timestamp。フォワードフィルデータの場合、これはソースの取り込み Timestamp です。バックフィルデータの場合、これは顧客によって提供されます。
record_source	STRING	ソース	"stream" (ライブストリームからのフォワードフィル) または "backfill" (バックフィルソースから)。
kafka_topic	STRING	Kafka	レコードが消費されたKafkaトピック。
kafka_partition	INT	Kafka	レコードが消費されたKafkaパーティション。
kafka_offset	LONG	Kafka	パーティション内のレコードのKafkaオフセット。

列	Type	ソース	説明
key	key_schemaによって異なります	ソース	指定したスキーマに従って構造化されたメッセージキー。
value	payload_schemaによって異なります	ソース	指定したスキーマに従って構造化されたメッセージ値（ペイロード）。
stream_record_timestamp	TIMESTAMP	ソース	レコードの Timestamp。フォワードフィルデータの場合、これはソースの取り込み Timestamp です。バックフィルデータの場合、これは顧客によって提供されます。
record_source	STRING	ソース	"stream" (ライブストリームからのフォワードフィル) または "backfill" (バックフィルソースから)。
kafka_topic	STRING	Kafka	レコードが消費されたKafkaトピック。
kafka_partition	INT	Kafka	レコードが消費されたKafkaパーティション。
kafka_offset	LONG	Kafka	パーティション内のレコードのKafkaオフセット。

バックフィルソース

フォワードフィルパイプラインはソースの最新の位置から起動するため、ストリームが作成される前に存在していたメッセージは取り込まれません。トレーニングのヒストリカルデータカバレッジを提供するには、オプションのバックフィルソースを設定します。

バックフィルソースが構成されている場合、Databricksは、バックフィル行をrecord_source="backfill"を使用して取り込みテーブルにコピーする1回限りのMERGE INTOジョブを実行します。MERGE は、重複チェッカーがバックフィルソースとフォワードフィルストリームに重複するタイムスタンプがあることを確認した後にのみ実行されます (バックフィルとライブストリームデータの重複を参照してください)。2日以内に重複条件が満たされない場合、無期限のブロックを回避するために、MERGEはとにかく実行されます。

バックフィルテーブルには、UTCタイムゾーンの型TIMESTAMPのstream_record_timestamp列が含まれている必要があります。その他のメタデータ列は、バックフィルソースに存在する場合はそのまま渡され、それ以外の場合はNULLに設定されます。Kafkaの場合、これらはkafka_topic、kafka_partition、およびkafka_offsetです。

Python

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

バックフィルとライブストリームデータの重複

バックフィルと取り込みテーブル間でMERGEを実行する前に、重複チェックが2つのテーブルのタイムスタンプを比較します：

• バックフィル最大値 ：バックフィルソースにおける最大stream_record_timestamp です。
• **取り込みの最小値**：stream_record_timestamp 取り込みテーブル内の行の最小record_source="stream" () です。

MERGEは、バックフィルの最新のタイムスタンプが取り込みテーブルの最も早いタイムスタンプを少なくとも1時間超えている場合に進行します。この重複により、取り込みテーブルにギャップが生じないようにします。重複条件が 2 日以内に満たされない場合でも、MERGE は無期限のブロックを回避するために実行されます。

取り込みパイプラインはソース内の最新の位置から起動するため、ストリームの作成後に到着したメッセージのみがキャプチャされます。バックフィルソースには、ストリームの作成時間までだけでなく、取り込みの時間範囲にまで及ぶデータが含まれている必要があります。

たとえば、午後3時にストリームを作成した場合、フォワードフィルパイプラインは午後3時以降のメッセージの読み取りを開始します。オーバーラップチェックを満たすには、バックフィルソースに少なくとも午後4時（フォワードフィル開始から1時間後）までのタイムスタンプ付きデータを含める必要があります。これは、取り込みテーブルにギャップがないことを確実にするため、午後4時以降にバックフィルテーブルを更新する必要があることを意味します。

重複排除

deduplication_columnsを使用して、バックフィルとフォワードフィルストリームデータの間の取り込み中に、重複する行を特定するための列パスを指定します。ネストされたフィールドにはドット表記を使用します（例："value.user_id"）。

データに基づいて重複排除カラムを選択します：

• ストリーム内の各レコードに一意の識別子 (例: value.transaction_id) が含まれている場合は、その列を重複排除に使用します。
• バックフィルソースにkafka_partitionとkafka_offsetカラムが含まれている場合は、それらを使用して各レコードを一意に識別します。
• 重複排除列が指定されていない場合、デフォルトの重複排除キーはkey、value、およびstream_record_timestampの完全な組み合わせです。これは推奨されません。この厳密な条件の一致は、簡単に重複を招く可能性があるためです。

Python

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

ストリームを管理

ストリームの取得

Python

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

ストリームを一覧表示

Python

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

include_schemas=Trueを設定して、完全なスキーマ詳細を含めます。スキーマは大規模になる可能性があり、これにより長時間実行される操作になることがあります。代わりにスキーマを個別に取得するには、get_stream を使用します。

ストリームの削除

ストリームを削除すると、その取り込みパイプラインおよび取り込みテーブルも削除されます。

警告

削除されたストリームを参照するモデルや特徴は、基になるストリームデータにアクセスできなくなります。このデータが必要だがストリームが不要になった場合は、削除する前に取り込みテーブルのコピーを作成してください。

Python

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

ノートブックの例

ストリームを作成し、ストリーミング特徴量を定義し、サービングエンドポイントにデプロイするエンドツーエンドの例については、以下のノートブックを参照してください。

ストリーミング特徴量ビュークイックスタートノートブック

ノートブックを新しいタブで開くOpen in Databricks