ストリームを設定する

Important

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

Stream は、Apache Kafka などの外部ストリーミングデータソースを表します。ストリームには、接続の詳細、認証、スキーマ、およびインジェスト構成が格納されます。ストリームが作成されたら、フィーチャビュー定義を使用してストリームを参照し、リアルタイムストリーミング機能を作成できます。

ストリームには 3 部構成の名前 (catalog.schema.stream_name) があります。ストリームへのアクセスは、関連するインジェストテーブルによって管理されます。詳細については、「インジェストとバックフィル」を参照してください。

必要条件

ノートブックコマンドを実行する場合: サーバーレスまたは Databricks Runtime 17.0 ML 以上を実行しているクラシックコンピューティングクラスター。
feature-engineering-client Python パッケージバージョン 0.16.0 以降をインストールする必要があります。

ストリームの作成

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

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

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    KafkaStreamConfig,
    KafkaSubscriptionMode,
    StreamConnectionConfig,
    DirectSchemas,
    SchemaConfig,
    IngestionConfig,
    IngestionDestination,
    StreamBackfillSource,
)

client = FeatureEngineeringClient()

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    source_config=KafkaStreamConfig(
        subscription_mode=KafkaSubscriptionMode(subscribe="events-topic"),
    ),
    connection_config=StreamConnectionConfig(
        uc_connection_name="my-kafka-connection"
    ),
    schema_config=DirectSchemas(
        payload_schema=SchemaConfig(
            json_schema=(
                '{'
                '  "type": "object",'
                '  "properties": {'
                '    "transaction_id": {"type": "string"},'
                '    "user_id": {"type": "string"},'
                '    "amount": {"type": "number"},'
                '    "event_time": {"type": "string", "format": "date-time"}'
                '  }'
                '}'
            )
        ),
    ),
    ingestion_config=IngestionConfig(
        ingestion_destination=IngestionDestination(
            delta_table_name="my_catalog.my_schema.events_ingestion"
        ),
    ),
)

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

ストリーミング機能を定義する前に、Kafka ブローカーにストリーミング Lakeflow Spark 宣言パイプラインのパイプライン接続を接続してテストします。サーバーレスコンピューティングでのストリーミングと Apache Kafka への接続を参照してください。

AWS マネージドストリーミング (Amazon MSK) については、「 Amazon MSK へのサーバーレスプライベート接続」を参照してください。 Kafka 認証オプションの詳細については、「認証」を参照してください。

Authentication

Unity カタログ接続 (推奨)

Unity カタログ接続を使用して、Kafka クラスターに対する認証を行います。これは、マネージド認証に推奨される方法です。接続を作成するには、「接続の作成」を参照してください。

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

直接 mTLS

直接 mTLS 認証の場合は、Databricks シークレットスコープを介して参照されるパスワードを使用して、Unity カタログボリュームに格納されているキーストアファイルとトラストストアファイルを提供します。 Kafka での SSL 認証の詳細については、「SSL を使用して kafka にAzure Databricks接続する」を参照してください。

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

プレビュー期間中、SASL 認証 (SASL/SCRAM と SASL/PLAIN の両方) はサポートされていません。

サブスクリプションモード

サブスクリプションモードでは、Stream が使用する Kafka トピックを選択する方法を指定します。次の 3 つのモードがサポートされています。

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

スキーマの構成

JSON スキーマ形式を使用して、メッセージキーと値の構造を定義します。 Kafka ソースの場合、 payload_schema は Kafka メッセージ値 (Kafka のキー値モデルの value ) に対応し、 key_schema は Kafka メッセージキーに対応します。少なくとも 1 つの payload_schema または 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"}'
    ),
)

キーまたはペイロードにスキーマが指定されていない場合は、単純な文字列として扱われます。

インジェストとバックフィル

ingestion_config パラメーターは、トレーニングとサービスのためにストリームデータをキャプチャして格納する方法を構成します。

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

SELECT インジェストテーブルで Stream への読み取りアクセスを許可します。
MANAGE インジェストテーブルで削除アクセスを許可します。

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

インジェストパイプライン

ストリームが作成されると、Databricks は、Kafka トピックからメッセージを継続的に読み取り、Delta テーブル (インジェストテーブル) に書き込むマネージドインジェストパイプラインを開始します。パイプラインは、最新の Kafka オフセットから開始され、継続的に実行され、ストリームの作成後に到着した新しいメッセージのみをキャプチャします。このインジェストテーブルは、ストリーミング機能を使用したトレーニングに使用されます。ストリームが削除されると、そのインジェストパイプラインとインジェストテーブルも削除されます。

取り込み先

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

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

インジェストテーブルスキーマ

インジェストテーブルには、メタデータ列と共にメッセージデータが含まれています。

Column	タイプ	Description
`key`	( `key_schema`によって異なります)	指定したスキーマに従って構造化された Kafka メッセージキー。
`value`	( `payload_schema`によって異なります)	指定したスキーマに従って構造化された Kafka メッセージ値 (ペイロード)。
`stream_record_timestamp`	`TIMESTAMP`	レコードのタイムスタンプ。前方入力データの場合、これは Kafka ブローカーの取り込みタイムスタンプです。バックフィルデータの場合、これは顧客が提供します。
`kafka_topic`	`STRING`	レコードが使用された Kafka トピック。
`kafka_partition`	`INT`	レコードが使用された Kafka パーティション。
`kafka_offset`	`LONG`	そのパーティション内における、そのレコードの Kafka オフセット。
`record_source`	`STRING`	`"stream"` (ライブ Kafka ストリームからの先行補完) または `"backfill"` (バックフィルソースから）

補完の供給元

前方フィルパイプラインは最新の Kafka オフセットから開始されるため、ストリームが作成される前に存在していたメッセージはキャプチャされません。トレーニング用の履歴データを確保するために、オプションのバックフィルソースを設定します。

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

バックフィルテーブルには、UTC タイムゾーンでstream_record_timestamp型のTIMESTAMP列を含める必要があります。他の Kafka メタデータ列 (kafka_topic、 kafka_partition、 kafka_offset) は、バックフィルソースに存在する場合はパススルーされるか、それ以外の場合は 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"
    ),
)

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

バックフィルとインジェストテーブルの間で MERGE を実行する前に、重複チェックによって 2 つのテーブルのタイムスタンプが比較されます。

バックフィルの最大値: バックフィルソース内の最大 stream_record_timestamp 。
取り込みの最小値: インジェストテーブル内の行数stream_record_timestamp（record_source="stream"）の最小値。

MERGE は、バックフィルの最新のタイムスタンプがインジェストテーブルの最も古いタイムスタンプを少なくとも 1 時間超えると続行されます。この重複により、インジェストテーブルにギャップがなくなります。重複条件が 2 日以内に満たされない場合は、MERGE が実行され、ブロックが無期限に回避されます。

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

たとえば、午後 3 時にストリームを作成した場合、前方入力パイプラインは午後 3 時以降にメッセージの読み取りを開始します。バックフィルソースには、重複チェックを満たすために、少なくとも午後 4:00 (前方入力開始から 1 時間) までのタイムスタンプを含むデータを含める必要があります。つまり、インジェストテーブルにギャップがないように、午後 4 時以降にバックフィルテーブルを更新する必要があります。

重複除去

deduplication_columnsを使用して、バックフィルストリームデータと前方フィルストリームデータ間のインジェスト中に重複する行を識別するための列パスを指定します。入れ子になったフィールド (たとえば、 "value.user_id") にはドット表記を使用します。

データに基づいて重複除去列を選択します。

ストリーム内の各レコードに一意の識別子 (たとえば、 value.transaction_id) が含まれている場合は、その列を重複除去に使用します。
バックフィルソースに kafka_partition 列と kafka_offset 列が含まれている場合は、それらを使用して各レコードを一意に識別します。
重複除去列が指定されていない場合、既定の重複除去キーは、 key、 value、および stream_record_timestampの完全な組み合わせです。この厳密な条件の一致によって重複が発生する可能性があり、これはお勧めしません。

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

ストリームの管理

ストリームを取得する

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

ストリームを一覧表示する

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

完全なスキーマの詳細を含むように include_schemas=True を設定します。スキーマは大きくなる可能性があり、実行時間の長い操作になる可能性があります。代わりにスキーマを個別に取得するには、 get_streamを使用します。

ストリームを削除する

ストリームを削除すると、そのインジェストパイプラインとインジェストテーブルも削除されます。

Warning

削除されたストリームを参照するモデルまたは機能は、基になるストリームデータにアクセスできなくなります。このデータが必要であってもストリームが不要になった場合は、削除前にインジェストテーブルのコピーを作成します。

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

ノートブックの例

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

ストリーミング機能ビューのクイックスタートノートブック

ノートブックを入手

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-06-26