ストリームを設定する

備考

プレビュー

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

Stream は、Apache Kafka や Amazon Kinesis などの外部ストリーミング データソースを表します。ストリームには、接続の詳細、認証、スキーマ、およびインジェストの設定が保存されます。ストリームの作成後、Feature View の定義を使用して参照し、リアルタイムのストリーミング特徴量を作成できます。

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

要件

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

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

ストリーミング特徴量を定義する前に、ストリーミング Lakeflow パイプライン接続を Kafka ブローカーまたは Kinesis サービス Endpoint に接続してテストします。Feature Store は Serverless SDP に依存しているため、クラシックコンピュート (ブローカーまたは Endpoint) を Databricks Serverless コンピュートに接続するメカニズムが必要です。これは、PrivateLink などの製品を使用するか、クラシックコンピュートがパブリックインターネットからアクセスできるようにすることで実行されます。

AWS マネージドストリーミング (Amazon MSK) については、Amazon MSK へのServerlessプライベート接続を参照してください。他のクラウドや Kinesis についても、同様のパターンが必要になります。

ストリームを作成

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

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

ソース固有の source_config と接続設定、および完全な create_stream() の例については、Apache Kafka または Amazon Kinesis を参照してください。スキーマとインジェストのオプションは、ソース間で共有されます。

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

Amazon Kinesis

Amazon Kinesis データストリームからストリームするには、ソース設定として KinesisStreamConfig を使用し、認証にはタイプ KINESIS の Unity Catalog 接続を使用します。

Python

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.entities import (
    KinesisStreamConfig,
    StreamNameList,
    StreamConnectionConfig,
    DirectSchemas,
    SchemaConfig,
    IngestionConfig,
    IngestionDestination,
)

client = FeatureEngineeringClient()

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    source_config=KinesisStreamConfig(
        stream_names=StreamNameList(names=["my-kinesis-stream"]),
    ),
    connection_config=StreamConnectionConfig(
        uc_connection_name="my-kinesis-connection"
    ),
    schema_config=DirectSchemas(
        payload_schema=SchemaConfig(
            json_schema=(
                '{'
                '  "type": "object",'
                '  "properties": {'
                '    "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"
        ),
    ),
)

Kinesis ストリーム識別子

読み込み元の Kinesis データストリームを特定するには、次のいずれかを正確に 1 つ使用します。1 つのストリームで複数の Kinesis データストリームから読み取ることができます。追加のソースオプションを extra_options 経由で渡します。たとえば、シャードごとの読み取りレートに上限を設定するための maxFetchRate や、default ポーリングコンシューマーの代わりに拡張ファンアウト (EFO) で読み取るための consumerMode="efo" などがあります。

フィールド	説明	例
stream_names	Kinesis ストリーム名のリスト	StreamNameList(names=["stream-a", "stream-b"])
stream_arns	Kinesis ストリーム ARN のリスト	StreamArnList(arns=["arn:aws:kinesis:us-west-2:123456789012:stream/stream-a"])

フィールド	説明	例
stream_names	Kinesis ストリーム名のリスト	StreamNameList(names=["stream-a", "stream-b"])
stream_arns	Kinesis ストリーム ARN のリスト	StreamArnList(arns=["arn:aws:kinesis:us-west-2:123456789012:stream/stream-a"])

Kinesis 認証

Kinesis は、Kinesis への読み取りアクセスを付与する IAM ロールである Unity Catalog サービス資格情報と、Stream の AWS リージョンを参照する、タイプ KINESIS の Unity Catalog 接続を介して認証を行います。接続を作成するには、Unity Catalog 接続を使用した認証を参照してください。ストリームの作成者は、接続に対して USE CONNECTION を持っている必要があります。ストリームをソースとして特徴量を実体化するユーザーも同様です。

SQL

CREATE CONNECTION IF NOT EXISTS `my-kinesis-connection`
TYPE KINESIS
OPTIONS (
    aws_region '<region>',
    credential '<service_credential>'
)

スキーマ構成

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

Kinesis ソースの場合、レコードデータには payload_schema のみを指定します。Kinesis メッセージのパーティションキーはスキーマのないルーティング文字列であるため、key_schema は適用されません。パーティションキーは、プレーン文字列として取り込みテーブルの key 列に引き続きキャプチャされます。

各 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 に設定します。

スキーマ進化

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

スキーマレジストリをサポートするストリームの場合、取り込みパイプラインは数時間ごとに自動的に再起動します。再起動するたびに、サブジェクトの最新のスキーマバージョンが取得され、新規または変更されたフィールドが取り込みテーブルに表示されます。

スキーマレジストリではなく直接スキーマを使用するストリームでは、update_streamを使用してスキーマが進化します。ストリームを更新を参照してください。

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

タイプ別にレコードを絞り込む

ストリームは、直接指定するかスキーマレジストリを使用するかに関わらず、（提供されている場合）単一のキーと値のスキーマを使用して各レコードをデコードします。トピックには複数のタイプのレコードを含めることができ、ストリームは複数のトピックをサブスクライブできるため、record_type_filter を使用して、トピック内のどのレコードがこのストリームに属するのかを選択します。

ドット表記でデコードされたフィールドを参照する SQL 式を指定します (例: value.event_type = 'transaction')。フィルターに一致しないレコードは無視されます。これらは取り込みテーブルには書き込まれず、マテリアライゼーションでも使用されません。他のレコード型のストリームを作成するには、異なる record_type_filter を指定して個別のストリームを作成します。

Python

stream = client.create_stream(
    name="my_catalog.my_schema.my_stream",
    # ...source, connection, schema, and ingestion config...
    record_type_filter="value.event_type = 'transaction'",
)

record_type_filter がなくても、デコードによってストリームが失敗することはありません。構成されたスキーマに一致しないレコードは、寛容にデコードされます。レコードは次のいずれかの方法でデコードされます。

• スキーマが想定しているもののレコードでは省略されているフィールドに対して、NULL 値を持つ行内へ (JSON、Avro、および Protobuf)。
• 異なるレコード型に属する値を保持する行へ (Avro および Protobuf のみ)。

インジェスト テーブル内の行が予期されるレコード型に属しているかどうかを識別するには、次のいずれかのチェックを使用します。

• フィールドが期待される値（例：value.event_type = 'transaction'）と等しいことを確認します（Avro と Protobuf に推奨）。
• フィールドが non-NULL であること（例：value.activity_id IS NOT NULL）を確認します。

トピック上のレコード型間でスキーマが大幅に異なる場合、または各レコード型へのアクセスを個別に管理したい場合は、個別のストリームと組み合わせて record_type_filter を使用することをお勧めします。コストを維持するために、各ストリームには個別のインジェストパイプラインとインジェストテーブルがあるため、Databricksでは少数のストリームを維持することをお勧めします。また、各ストリームはマテリアライズ時に個別のコンピュートを使用します。マテリアライズ用に機能固有のフィルターを使用できます。

record_type_filter は、特徴量の filter_condition とは異なります。record_type_filter はストリーム上で設定され、どのレコードが取り込まれ、ストリームをソースとして使用するすべての特徴量で使用可能になるかを制御し、一方 filter_condition は個々の特徴量上で設定され、集計前に行をフィルタリングします。filter_condition の詳細については、ストリーミングソースのフィルター条件を参照してください。

取り込みとバックフィル

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 ストリームにのみ、kinesis_* の列は Kinesis ストリームにのみ存在します。

列	Type	ソース	説明
key	key_schemaによって異なります	ソース	指定したスキーマに従って構造化されたメッセージキー。Kinesis ストリームにはキーのスキーマがないため、これは STRING としてのレコードのパーティションキーです。
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オフセット。
kinesis_stream	STRING	Kinesis	レコードのコンシューミード元となった Kinesis データストリーム。
kinesis_shard_id	STRING	Kinesis	レコードのコンシューマー元となったシャード。
kinesis_sequence_number	STRING	Kinesis	シャード内のレコードのシーケンス番号。

列	Type	ソース	説明
key	key_schemaによって異なります	ソース	指定したスキーマに従って構造化されたメッセージキー。Kinesis ストリームにはキーのスキーマがないため、これは STRING としてのレコードのパーティションキーです。
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オフセット。
kinesis_stream	STRING	Kinesis	レコードのコンシューミード元となった Kinesis データストリーム。
kinesis_shard_id	STRING	Kinesis	レコードのコンシューマー元となったシャード。
kinesis_sequence_number	STRING	Kinesis	シャード内のレコードのシーケンス番号。

バックフィルソース

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

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

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

Kinesis の場合、パススルーメタデータ列は kinesis_stream、kinesis_shard_id、および kinesis_sequence_number です。

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の完全な組み合わせです。これは推奨されません。この厳密な条件の一致は、簡単に重複を招く可能性があるためです。

Kinesis ストリームの場合は、重複排除のために kinesis_shard_id と kinesis_sequence_number を組み合わせて使用します（シーケンス番号はシャード内でのみ一意です）。さらに、Stream が複数の Kinesis ストリームを読み取る場合は kinesis_stream も使用します。

Python

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

コストアトリビューション

Streamの管理された取り込みのコストを属性に割り当てるには、IngestionConfigでtagsとbudget_policy_idを設定します。Databricksは、Streamの作成時に、それらを取り込みLakeflowパイプライン、およびそのフォワードフィルとバックフィルのジョブに適用します。

例、タグの制限、および属性された支出のクエリー方法については、タグとServerless使用ポリシーを使用したコストの属性を参照してください。

ストリームから列を除外する

取り込みたくない特定の列をストリームから削除するには、excluded_columnsを使用します。除外された列は取り込みテーブルに書き込まれず、特徴量で参照したり、トレーニングで使用したりすることはできません。

value.user.email や key.account_id などのメッセージ キーまたは値へのドット表記を使用して、各列を指定します。これらの列は、取り込み、バックフィル、およびマテリアライズ全体で、デコードされた key および value から削除されます。パスが構造体を指している場合、そのすべてのネストされたフィールドも削除されます（たとえば、value.address では value.address.city と value.address.zip も削除されます）。

key.* パスにはメッセージキーを持つソースが必要です。Amazon Kinesis ストリームにはメッセージキーがないため、Kinesis ストリームは value.* の除外のみをサポートします。

Python

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

直接スキーマを使用する場合、除外された列はキーまたは値のスキーマに既に存在している必要があります。存在しない場合、create_stream は失敗します。スキーマレジストリを使用する場合、存在しなくなる前に列を除外できます。また、重複排除列は重複行を特定するために必須であるため、除外された列を重複排除列にすることはできません。除外された列を参照するフィーチャ（たとえば、エンティティ、時系列、入力などとして）は、作成に失敗します。

update_stream直接スキーマおよびスキーマレジストリバックのストリームの両方で、作成後にストリームの除外列を変更できます。詳細については、ストリームの更新を参照してください。

ストリームを管理

ストリームの取得

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 を使用します。

ストリームを更新

作成後にストリームを変更するには、update_stream を使用します。直接スキーマを進化させるには schema_config を渡し、削除する列を変更するには excluded_columns を渡すか、あるいはその両方を渡します。他のフィールドの更新はサポートされていません。代わりに、新しいストリームを作成してください。

ストリームを更新すると、変更を有効にするために取り込みパイプラインが再起動します。取り込みは通常、数分以内に再開します。

ダイレクトスキーマを進化させる

直接スキーマを使用するストリームの場合は、DirectSchemas を schema_config に渡します。payload_schema、key_schema、またはその両方を設定します。設定していない側は変更されません。スキーマレジストリをバックアップとするストリームは、schema_config の更新を拒否するため、代わりにレジストリを通じて進化させる必要があります。

Python

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

実行中の取り込みパイプラインが既存のレコードのデコードと取り込みテーブルへの書き込みを継続できるように、スキーマの更新は下位互換性を持つ必要があります。その他の変更は拒否されます。

許可される内容はフォーマットによって異なります：

• JSON および Protobuf : オプションフィールドの追加、フィールドの削除、およびフィールドの型の拡大（例: int から bigint への変更）を行います。また、Protobuf ではフィールドの並び替えも可能です。
• Avro : int を long に広げること、および、後続のフィールドがバイトを読み取らない末尾のフィールドを削除することのみを許可します。Avroスキーマをより柔軟に進化させるには、代わりにスキーマレジストリでサポートされたストリームを使用します。

フィールドを追加すると、取り込みテーブルのデコードされた key および value 構造体が大きくなります。更新前に書き込まれた行はその元の形状を維持し、それ以前の行では追加されたフィールドが NULL として読み込まれます。削除および型の変更は、更新後に取り込まれたレコードに対してのみ有効になります。

除外する列を変更する

列パスの完全な新しいセットを excluded_columns に渡し、既存のセットを置き換えます。すべての除外をクリアするには、空のリスト（[]）を渡します。この動作の詳細については、ストリームからの列の除外を参照してください。

Python

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

除外する列の変更は前方のみです。新しく除外された列の書き込みが停止し（NULL に表示）、新しく含まれた列のデータ入力が今後に向けて起動されますが、以前に書き込まれた行はそのまま残されます。新しい列の取り込みを常に防ぐには：

• スキーマレジストリ ：まず列を excluded_columns に追加し、取り込みパイプラインの再起動を待ってから、レジストリに新しいスキーマバージョンを登録します。
• 直接スキーマ ：同じupdate_stream呼び出しで、schema_configとexcluded_columnsに列を追加します。

ストリームの削除

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

警告

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

Python

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

ノートブックの例

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

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

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