스트림 설정

스트림은 Apache Kafka와 같은 외부 스트리밍 데이터 원본을 나타냅니다. 스트림은 연결 세부 정보, 인증, 스키마 및 수집 구성을 저장합니다. 스트림이 생성된 후에는 Feature View 정의에서 이를 참조하여 실시간 스트리밍 기능을 생성할 수 있습니다.

스트림에는 세 부분으로 구성된 이름(catalog.schema.stream_name)이 있습니다. Stream에 대한 액세스는 연결된 수집 테이블에 의해 제어됩니다. 자세한 내용은 섭취 및 백필 을 참조하세요.

요구 사항

• Notebook 명령을 실행하는 경우: Databricks Runtime 17.0 ML 이상을 실행하는 서버리스 또는 클래식 컴퓨팅 클러스터입니다.
• feature-engineering-client Python 패키지 버전 0.18.0 이상이 반드시 설치되어야 합니다.

스트림 소스에 연결

스트리밍 기능을 정의하기 전에 Kafka broker에 대한 스트리밍 Lakeflow 파이프라인 연결을 연결하고 테스트합니다. Feature Store는 서버리스 SDP에 의존하기 때문에, 클래식 컴퓨트(브로커 또는 엔드포인트)를 Databricks 서버 리스 컴퓨트에 연결할 수 있는 메커니즘이 필요합니다. 이는 privatelink 같은 제품을 통해서, 또는 공용 인터넷에서 클래식 컴퓨트에 접근할 수 있도록 하는 방식으로 이루어집니다.

스트림 만들기

새 스트림을 만드는 데 사용합니다 create_stream() . 스트림에는 다음 네 가지 구성 요소가 필요합니다.

• 소스 설정: 스트리밍 플랫폼과 Kafka 소스의 주제 구독 등 소스별 세부 정보를 지정합니다.
• 연결 구성: 부트스트랩 서버 및 자격 증명을 포함하여 스트리밍 플랫폼에 연결하고 인증하는 방법을 지정합니다.
• 스키마 구성: 메시지 키 및 값의 구조를 정의합니다.
• 수집 구성: 스트림 데이터가 수집되는 위치와 방법을 지정합니다. 자세한 내용은 섭취 및 백필 을 참조하세요.

소스별 source_config 및 연결 설정과 전체 create_stream() 예제는 Apache Kafka를 참조하세요. 스키마와 인제션 옵션은 소스마다 공유됩니다.

아파치 카프카

Apache Kafka에서 스트리밍하려면 소스 설정으로 사용하고 KafkaStreamConfig 인증을 위한 Unity 카탈로그 연결을 사용하세요. 서버리스 컴퓨팅에서의 스트리밍 및 Kafka 연결에 대해서는 Apache 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 구독 모드

구독 모드는 Stream에서 사용할 Kafka 토픽을 선택하는 방법을 지정합니다. 세 가지 모드가 지원됩니다.

모드	Description	예시
subscribe	토픽 이름의 쉼표로 구분된 목록	KafkaSubscriptionMode(subscribe="topic1,topic2")
subscribe_pattern	Java 정규식 패턴과 일치하는 토픽 이름	KafkaSubscriptionMode(subscribe_pattern="events-.*")
assign	토픽 파티션 할당을 지정하는 JSON	KafkaSubscriptionMode(assign='{"my-topic": [0, 1, 2]}')

카프카 인증

Unity 카탈로그 연결(권장)

Unity 카탈로그 연결을 사용하여 Kafka 클러스터에 인증합니다. 관리 인증에 권장되는 방법입니다. 연결을 만들려면 연결 만들기를 참조하세요. 스트림의 생성자는 연결에 USE CONNECTION이(가) 있어야 합니다. Stream을 소스로 사용하는 피처를 구체화하는 모든 사용자는 해당 연결에서 USE CONNECTION도 반드시 가지고 있어야 합니다.

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

이 연결은 IAM(서비스 자격 증명)과 SASL 인증을 모두 지원합니다.

IAM (서비스 자격 증명)

예를 들어, IAM 사용으로 Amazon MSK에 연결하려면 Unity Catalog 서비스 자격 증명으로 인증하세요. 서비스 자격 증명을 만들려면 서비스 자격 증명 만들기를 참조하세요. 서비스 자격 증명 이름을 다음과 같은 옵션으로 credential 설정하세요:

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

자격 증명서와 userpassword 옵션을 제공하세요. 이 연결은 이러한 자격 증명을 안전하게 저장합니다.

다음 예시는 SASL/SCRAM을 사용합니다. SASL/PLAIN의 경우 sasl_mechanism을(를) 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>'
)

직접 mTLS

직접 mTLS 인증을 위해 Unity Catalog 볼륨에 저장된 키스토어 및 트러스트스토어 파일을 제공하세요. 비밀번호는 Databricks 시크릿 스코프를 통해 참조되어야 합니다. 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"
        ),
    ),
)

스키마 구성

메시지 키와 값의 구조를 정의하여 인제스트와 기능 정의가 개별 필드를 읽을 수 있도록 하세요. Kafka 원본의 payload_schema 경우 Kafka 메시지 값( value Kafka의 키-값 모델)에 해당하며 key_schema Kafka 메시지 키에 해당합니다. payload_schema 또는 key_schema 중 하나 이상은 반드시 제공되어야 합니다.

각 SchemaConfig 메시지는 소스가 메시지를 직렬화하는 방식에 맞춰 세 가지 형식 중 하나를 수용합니다: json_schema, avro_schema, 또는 proto_schema. 키 또는 페이로드에 대해 스키마가 제공되지 않으면 간단한 문자열로 처리됩니다.

이 섹션의 코드 예제에서는 스키마가 문자열로 제공되는 DirectSchemas와 함께 인라인으로 선언된 스키마를 사용합니다. 외부 스키마 레지스트리를 사용해 스키마를 관리하려면 스키마 레지스트 리를 참조하세요.

JSON 스키마

에 JSON 스키마 문자열 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"}'
    ),
)

Avro 스키마

avro_schema에 Avro 스키마 문자열을 제공하세요. Avro 논리 타입은 지원되며, 여기에는 timestamp-millis, date, 그리고 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"}}'
            '  ]'
            '}'
        )
    ),
)

Protobuf 스키마

.proto에 proto_schemaProtoSchemaSpec 소스 텍스트와 페이로드 메시지 이름이 포함되도록 제공합니다. ProtoSchemaSpec에서 databricks.feature_engineering.entities을 가져옵니다.

message_name는 package 텍스트에 선언된 .proto를 포함한 정규화된 전체 메시지 이름이어야 합니다(예: com.example.Event, Event 아님). proto2와 proto3 문법 모두 지원됩니다.

google.protobuf.Timestamp 스칼라 래퍼 유형(StringValue, Int32Value, 등)이 지원되며, 이들의 가져오기는 자동으로 해결됩니다. , 와 같은 다른 잘 알려진 타입DurationStructAny은 거부되며; 대신 해당 값들을 지원되는 스칼라 또는 메시지로 인코딩합니다. fixed32 및 fixed64 스칼라 형식과 문자열이 아닌 키를 사용하는 map도 지원되지 않습니다.

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 유니온. 여러 레코드 유형의 유니언은 각 레코드 유형당 하나의 필드를 가진 구조체로 디코딩되며, 각 필드는 Avro 레코드 이름을 따서 명명됩니다.
• 프로토부프 타입들. 부호 없는 정수는 더 넓은 부호 있는 타입으로 디코딩되며(예: DECIMAL(20,0)은(는) StringValue로, Int32Value은(는) uint64로), enum 필드는 해당 문자열 이름으로 디코딩되고, 스칼라 래퍼 타입(예: BIGINT 및 uint32)은 래핑된 타입의 nullable 열로 디코딩됩니다.

스키마 레지스트리

스키마 레지스트리는 스트리밍 제작자와 소비자가 사용하는 버전 스키마를 저장하며, 스키마가 발전함에 따라 호환성 규칙을 강제합니다. 외부 스키마 레지스트리가 설정되면, Feature Store는 레지스트리에서 스키마를 읽어 스트리밍 메시지를 디코딩합니다. 스키마 레지스트리를 사용할 때는 스트림에서 스키마를 인라인으로 선언하지 않습니다.

스키마 레지스트리 지원에는 다음과 같은 제한이 있습니다:

• Kafka 스트림에만 지원됨.
• Confluent 스키마 레지스트리만 지원됩니다
• 지원되는 것은 Avro 와 Protobuf 포맷뿐입니다. JSON 메시지를 읽으려면 스키마를 인라인으로 선언하세요. JSON 스키마를 참고하세요.
• 각 스트림은 메시지 값에 대해 정확히 하나의 Confluent subject에 연결되며, 메시지 키에 대해서도(제공된 경우) 정확히 하나의 Confluent subject에 연결됩니다. 여러 스키마 레코드를 포함하는 스트림 토픽은 지원되는 구성이 아닙니다. 스트림이 여러 스키마를 포함하는 주제에 연결된다면, 지정된 주제의 스키마와 일치하지 않는 레코드는 null로 디코딩됩니다.

스키마 레지스트리에 연결하세요

Kafka Unity 카탈로그 연결에서 레지스트리 연결 세부 정보를 옵션으로 제공하고, 레지스트리 API 비밀을 Databricks 비밀 범위에 저장하세요. 스트림의 run-as ID에는 시크릿 범위에 대한 READ 권한이 있어야 합니다. 수집 파이프라인이 런타임에 시크릿을 읽기 때문입니다. 연결을 만들고 설정하는 방법에 대해서는 연결 생성(Create a connection)을 참조하세요.

인증에 사용되는 연결에 , , schema_registry_api_key, 그리고 schema_registry_api_secret 옵션을 추가schema_registry_url하세요. 다음 예시는 Unity Catalog 서비스 자격 증명으로 브로커에게, API 키로 레지스트리에 인증하는 Kafka 연결을 생성합니다:

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 연결의 옵션과 Stream의 schema_registry_api_secret 참조를 모두 동일한 시크릿으로 설정하세요.

스키마 레지스트리를 사용하는 스트림을 생성하세요

SchemaRegistryConfig를 schema_config로 전달하세요. api_secret_ref로 레지스트리 API 시크릿을 참조하고, 메시지 값의 경우 key_schema_locator로, 메시지 키의 경우 payload_schema_locator로 주제와 형식을 식별합니다. 최소 한 개의 위치 추적기가 제공되어야 합니다.

Schema configuration 섹션의 직접적인 스키마 예시와 비교했을 때 여기에서의 차이점에 주목하세요. 스키마 레지스트리를 사용할 때는 스트림 schema_config에 스키마를 인라인으로 제공하지 않습니다. 대신, 레지스트리에서 스키마를 식별하는 a SchemaRegistryConfig 를 지정합니다.

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(기본값, 토픽 이름에서 제목을 파생): 값의 경우 <topic>-value, 키의 경우 <topic>-key. 예를 들어, 주제 transactions 의 값 스키마는 주어 transactions-value를 사용합니다.
• RecordNameStrategy (토픽과 무관하게 스키마의 레코드 이름에서 subject를 결정): 완전 수식된 레코드 이름(예: 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를 사용하여 해당 토픽의 레코드 중 어떤 레코드가 이 스트림에 해당하는지 선택하세요.

예를 들어 value.event_type = 'transaction'와 같이 점 표기법을 사용해 디코딩된 필드를 참조하는 SQL 표현식을 제공합니다. 필터와 일치하지 않는 레코드는 무시됩니다. 수집 테이블에 기록되지 않으며, 구체화에도 사용되지 않습니다. 다른 레코드 유형용 스트림을 만들려면 다른 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'",
)

record_type_filter가 없어도 디코딩으로 인해 스트림이 절대 실패하지 않습니다. 설정된 스키마와 일치하지 않는 레코드는 관대하게 디코딩됩니다. 기록은 다음과 같은 방식으로 디코딩됩니다:

• 스키마가 기대하지만 레코드에서 누락된 필드에 대한 값(JSON, Avro, Protobuf)을 포함하는 NULL 행
• 다른 레코드 유형의 값이 포함된 행으로(Avro 및 Protobuf에만 해당)

수집 테이블의 행이 예상되는 레코드 유형에 속하는지 확인하려면 다음 확인 방법 중 하나를 사용하세요:

• 예를 들어 value.event_type = 'transaction'와 같이 필드가 예상 값과 같은지 확인합니다(Avro 및 Protobuf에 권장).
• 필드가 NULL이 아닌지 확인하세요. 예를 들어 value.activity_id IS NOT NULL.

토픽의 레코드 유형 간 스키마 차이가 크거나 각 레코드 유형에 대한 액세스를 개별적으로 관리하려는 경우에는 별도의 Streams와 함께 record_type_filter를 사용하는 것이 권장됩니다. 비용을 관리하기 위해 Databricks는 각 스트림마다 별도의 인제스 파이프라인과 인제스 테이블이 있으므로 소수의 스트림을 유지할 것을 권장합니다. 각 스트림은 구체화 시점에 별도의 컴퓨트도 사용합니다. 특성별 필터를 사용해 머티리얼라이제이션할 수 있습니다.

record_type_filter는 기능의 filter_condition와 다릅니다. record_type_filter는 스트림에 설정되어 스트림을 소스로 사용하는 모든 기능에 어떤 레코드가 수집되고 접근 가능한지 제어하는 반면 filter_condition는 개별 기능에 설정되어 집계 전에 행을 필터링합니다. 에 대한 자세한 내용은 filter_condition을 참조하세요.

섭취 및 되채움

매개 변수는 ingestion_config 학습 및 서비스를 위해 스트림 데이터를 캡처하고 저장하는 방법을 구성합니다.

스트림에 대한 액세스는 수집 테이블에 의해 제어됩니다.

• SELECT 수집 테이블에서 Stream에 대한 읽기 권한을 부여합니다.
• 수집 테이블의 MANAGE는 삭제 권한을 부여합니다.

테이블 권한에 대한 자세한 내용은 테이블 및 Unity 카탈로그 권한 참조를 참조하세요.

데이터 수집 파이프라인

스트림이 생성되면 Databricks는 관리자형 인제스션 파이프라인을 시작하여 소스 스트림에서 메시지를 지속적으로 읽어 델타 테이블(인제스션 테이블)에 씁니다. 파이프라인은 소스 내 최신 위치에서 시작하여 스트림이 생성된 후 도착하는 새로운 메시지만 계속 캡처하며 계속 실행됩니다. 이 수집 테이블은 스트리밍 기능을 사용하여 학습하는 데 사용됩니다. 스트림이 삭제되면 해당 수집 파이프라인 및 수집 테이블도 삭제됩니다.

수집 대상

스트림 ingestion_destination 데이터가 기록되는 세 부분으로 구성된 델타 테이블 이름을 지정합니다.

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

수집 테이블 스키마

인제스션 테이블에는 메시지 데이터와 메타데이터 열이 포함되어 있습니다. 공통 열은 모든 소스에 있으며, kafka_* 열은 Kafka 스트림에만 있습니다.

Column	Type	Source	Description
key	출발지(key_schema)에 따라 달라집니다	Common	메시지 키는 당신이 제공한 스키마에 따라 구조화되어 있습니다.
value	출발지(payload_schema)에 따라 달라집니다	Common	메시지 값(페이로드)은 제공하신 스키마에 따라 구조화되어 있습니다.
stream_record_timestamp	TIMESTAMP	Common	기록의 타임스탬프입니다. 순방향 채우기 데이터의 경우, 이 데이터가 소스 인제스트 타임스탬프입니다. 백필 데이터는 고객이 제공한 데이터입니다.
record_source	STRING	Common	"backfill"(라이브 스트림에서 포워드 필) 또는 "stream"(백필 소스에서) 중 하나입니다.
kafka_topic	STRING	Kafka	레코드가 소비된 Kafka 토픽입니다.
kafka_partition	INT	Kafka	레코드가 소비된 Kafka 파티션입니다.
kafka_offset	LONG	Kafka	해당 파티션 내 레코드의 카프카 오프셋입니다.

백필 소스

순방향 채우기 파이프라인은 소스의 최신 위치에서 시작되기 때문에 스트림이 생성되기 전에 존재했던 메시지를 캡처하지 않습니다. 학습용 과거 데이터 범위를 제공하려면 선택적 백필 소스를 구성하세요.

백필 소스가 구성되면 Databricks는 백필 행을 MERGE INTO와 함께 인제스트 테이블에 복사하는 일회성 record_source="backfill" 작업을 실행합니다. MERGE는 중첩 검사기가 백필 원본과 앞으로 채우기 스트림에 겹치는 타임스탬프가 있음을 확인한 후에만 실행됩니다( 백필과 라이브 스트림 데이터 간의 겹침 참조). 중첩 조건이 2일 이내에 충족되지 않더라도 무기한 차단을 방지하기 위해 MERGE는 계속 실행됩니다.

백필 테이블에는 UTC 시간대의 stream_record_timestamp 유형인 TIMESTAMP 열이 포함되어야 합니다. 다른 메타데이터 열은 백필 소스에 있으면 그대로 전달되며, 그렇지 않으면 NULL로 설정됩니다. 카프카의 경우, 이들은 , kafka_partition, 입니다kafka_offset.kafka_topic

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를 실행하기 전에 중복 확인을 위해 두 테이블의 타임스탬프를 비교합니다.

• 백필 최대값: 백필 소스의 최대 stream_record_timestamp 값입니다.
• 수집 최소: 수집 테이블의 최소 stream_record_timestamp 행(record_source="stream")입니다.

MERGE는 백필의 최신 타임스탬프가 수집 테이블의 가장 빠른 타임스탬프를 1시간 이상 초과하면 진행됩니다. 이 겹침으로 인해 수집 테이블에 공백이 생기지 않습니다. 중첩 조건이 2일 이내에 충족되지 않더라도 무기한 차단을 방지하기 위해 MERGE는 계속 실행됩니다.

인제스팅 파이프라인은 소스의 최신 위치에서 시작하기 때문에 스트림이 생성된 후에 도착한 메시지만 캡처합니다. 백필 원본에는 스트림 생성 시간뿐만 아니라 수집 시간 범위로 확장되는 데이터가 포함되어야 합니다.

예를 들어 오후 3시에 스트림을 만드는 경우 전달 채우기 파이프라인은 오후 3시 이후부터 메시지를 읽기 시작합니다. 겹침 검사를 충족하려면 백필 소스에 최소 오후 4:00까지의 타임스탬프가 포함된 데이터(포워드필 시작 시점보다 1시간 후)가 포함되어 있어야 합니다. 즉, 수집 테이블에 누락 구간이 없도록 오후 4시 이후에 백필 테이블을 업데이트해야 합니다.

Deduplication

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

비용 배분

IngestionConfig에서 tags 및 budget_policy_id을 설정하여 스트림의 관리형 수집 비용을 귀속하세요. Azure Databricks는 스트림이 생성될 때 이를 수집 Lakeflow 파이프라인과 해당 전방 채우기 및 백필 작업에 적용합니다.

예를 들어, 태그 한도와 귀속된 지출을 조회하는 방법은 ' 태그와 서버리스 사용 정책이 포함된 속성 비용'을 참고하세요.

스트림에서 열을 제외하기

수집하지 않으려는 특정 열을 스트림에서 드롭하는 데 excluded_columns를 사용하세요. 제외된 열은 수집 테이블에 저장되지 않으며 피처에서 참조되거나 학습에 사용될 수 없습니다.

각 열을 메시지 키 또는 값에 점 표기법으로 지정하세요. 예: value.user.email 또는 key.account_id. 이 열들은 디코딩된 key와 value에서 인제션, 백필, 물질화 전반에 걸쳐 제거됩니다. 경로가 구조체를 가리키면, 그 구조체의 모든 중첩된 필드도 함께 삭제됩니다(예를 들어, value.address도 value.address.city 및 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"],
)

직접 스키마를 사용할 때, 제외된 열은 이미 키 또는 값 스키마에 존재해야 하며, 그렇지 않으면 create_stream이(가) 실패합니다. 스키마 레지스트리를 사용할 때는 열이 존재하기 전에 열을 제외할 수 있습니다. 제외된 열은 중복 제거 열이 될 수 없는데, 중복 제거 열은 중복 행을 식별하는 데 필수이기 때문입니다. 제외된 열을 참조하는 모든 기능(예를 들어 엔터티, 시계열 또는 입력으로 참조하는 경우)은 생성할 수 없습니다.

직접 스키마를 사용하는 스트림과 스키마 레지스트리를 사용하는 스트림 모두에서, 스트림의 제외 열은 생성 후 update_stream로 변경할 수 있습니다. 자세한 내용은 Update a stream을(를) 참조하세요.

스트림 관리

스트림 가져오기

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.

스트림 업데이트

스트림을 생성한 후 변경하는 데 update_stream를 사용하세요. 직접 스키마를 진화시키려면 excluded_columns를 전달하고, 삭제할 열을 변경하려면 schema_config를 전달하거나, 둘 다 전달하세요. 다른 필드는 업데이트할 수 없습니다. 대신 새 스트림을 만들어 보세요.

스트림을 업데이트하면 수집 파이프라인이 재시작되어 변경 사항이 적용됩니다. 데이터 수집은 보통 몇 분 내에 재개됩니다.

직접 스키마 진화

직접 스키마를 사용하는 스트림의 경우, schema_config에 DirectSchemas를 전달합니다. payload_schema, key_schema, 또는 둘 다를 설정합니다. 설정하지 않은 쪽은 변하지 않습니다. 스키마 레지스트리 지원 스트림은 업데이트 schema_config를 거부하며 대신 레지스트리를 통해 변경해야 합니다.

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: long을 int로 확장하는 것과 뒤따르는 어떤 필드도 그 바이트를 읽지 않는 후행 필드를 제거하는 것만 허용합니다. Avro 스키마를 더 자유롭게 발전시키려면 대신 스키마 레지스트리 기반 스트림을 사용하세요.

필드를 추가하면 인제스션 테이블이 디코딩 keyvalue 되고 구조체가 커집니다. 업데이트 이전에 작성된 행들은 원래 형태를 유지하며, 추가된 필드는 이전 행들과 동일 NULL 하게 읽힙니다. 삭제 및 유형 변경은 업데이트 이후 인제스트된 레코드에만 적용됩니다.

제외된 열 변경

새로운 열의 경로 excluded_columns집합을 에 전달하여 기존 집합을 대체합니다. 모든 제외 항목을 제거하려면 빈 목록([])을 전달합니다. 이 동작에 대한 자세한 내용은 스트림에서 열 제외하기를 참조하세요.

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

제외된 열을 변경하는 것은 앞으로만 가능합니다. 새로 제외된 열은 더 이상 쓰이지 않고(에 나타나며 NULL), 새로 포함된 열은 앞으로 채워지기 시작하며, 이전에 작성된 행은 as-is남게 됩니다. 새로운 칼럼이 절대 인지되지 않도록 방지하려면:

• 스키마 레지스트리: 열을 excluded_columns 먼저 추가하고 인제스팅 파이프라인이 재시작될 때까지 기다린 후 새 스키마 버전을 레지스트리에 등록하세요.
• 직접 스키마: 같은 update_stream 호출 내에서 와 schema_config 에 excluded_columns 열을 추가합니다.

스트림 삭제

스트림을 삭제하면 해당 수집 파이프라인 및 수집 테이블도 삭제됩니다.

Warning

삭제된 스트림을 참조하는 모든 모델 또는 기능은 더 이상 기본 스트림 데이터에 액세스할 수 없습니다. 이 데이터가 필요하지만 스트림이 더 이상 필요하지 않은 경우 삭제하기 전에 수집 테이블의 복사본을 만듭니다.

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

예제 노트

Stream을 만들고, 스트리밍 기능을 정의하고, 서비스 엔드포인트에 배포하는 엔드 투 엔드 예제는 다음 Notebook을 참조하세요.

스트리밍 기능 보기 빠른 시작 노트북

노트북 받기