스트림 설정

중요합니다

이 기능은 공개 미리보기 단계에 있습니다. 작업 영역 관리자는 미리 보기 페이지에서 이 기능에 대한 액세스를 제어할 수 있습니다. Azure Databricks 미리 보기 관리를 참조하세요.

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

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

요구 사항

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

스트림 만들기

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

• 원본 구성: 스트리밍 플랫폼(예: 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 broker에 대한 스트리밍 Lakeflow 파이프라인 연결을 연결하고 테스트합니다. 서버리스 컴퓨팅에서 스트리밍을 참조하고 Apache Kafka에 연결합니다.

AWS 관리 스트리밍(Amazon MSK)의 경우 Amazon MSK에 대한 서버리스 프라이빗 연결을 참조하세요. Kafka 인증 옵션에 대한 자세한 내용은 인증을 참조 하세요.

Authentication

Unity 카탈로그 연결(권장)

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

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

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

SASL

SASL 인증(SASL/SCRAM 및 SASL/PLAIN 모두)은 미리 보기 중에 지원되지 않습니다.

구독 모드

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

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

스키마 구성

메시지 키와 값의 구조를 정의하여 인제스트와 기능 정의가 개별 필드를 읽을 수 있도록 하세요. 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는 레지스트리에서 스키마를 읽어 스트리밍 메시지를 디코딩합니다. 스키마 레지스트리를 사용할 때는 스트림에서 스키마를 인라인으로 선언하지 않습니다.

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

• 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로 설정하세요.

스키마 진화

섭취 파이프라인은 시작될 때 피험자의 현재 스키마를 해결합니다. 주제에 대해 스키마 레지스트리에 새로운 하위 호환 스키마 버전을 등록하면, 실행 중인 파이프라인은 처음 사용하던 버전을 계속 사용합니다.

Databricks는 인제스팅 파이프라인을 서버리스 Lakeflow 파이프라인으로 관리하기 때문에 파이프라인은 주기적으로 재시작됩니다. 다음 재시작 시 새로운 스키마 버전을 선택합니다. 입력 테이블에 새로운 필드나 변경된 필드가 나타나기까지 최대 일주일이 걸릴 수 있습니다.

파이프라인이 현재 사용 중인 스키마와 일치하지 않는 레코드를 어떻게 처리하는지에 대해서는 '스키마를 이용한 데이터 디코딩'을 참조하세요.

수집 및 과거 데이터 채우기

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

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

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

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

데이터 수집 파이프라인

스트림이 생성되면 Databricks는 Kafka 토픽에서 메시지를 지속적으로 읽어 Delta 테이블(즉, 수집 테이블)에 기록하는 관리형 수집 파이프라인을 시작합니다. 파이프라인은 최신 Kafka 오프셋에서 시작하여 지속적으로 실행되어 스트림을 만든 후에 도착하는 새 메시지만 캡처합니다. 이 수집 테이블은 스트리밍 기능을 사용하여 학습하는 데 사용됩니다. 스트림이 삭제되면 해당 수집 파이프라인 및 수집 테이블도 삭제됩니다.

수집 대상

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

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

수집 테이블 스키마

수집 테이블에는 메타데이터 열과 함께 메시지 데이터가 포함됩니다.

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

백필 소스

정방향 채우기 파이프라인은 최신 Kafka 오프셋에서 시작되므로 스트림을 만들기 전에 존재했던 메시지를 캡처하지 않습니다. 학습용 과거 데이터 범위를 제공하려면 선택적 백필 소스를 구성하세요.

백필 소스가 구성되면 Databricks는 백필 행을 MERGE INTO와 함께 인제스트 테이블에 복사하는 일회성 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를 실행하기 전에 중복 확인을 위해 두 테이블의 타임스탬프를 비교합니다.

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

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

수집 파이프라인은 최신 Kafka 오프셋에서 시작되므로 스트림을 만든 후에 도착하는 메시지만 캡처합니다. 백필 원본에는 스트림 생성 시간뿐만 아니라 수집 시간 범위로 확장되는 데이터가 포함되어야 합니다.

예를 들어 오후 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"],
)

스트림 관리

스트림 가져오기

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을 만들고, 스트리밍 기능을 정의하고, 서비스 엔드포인트에 배포하는 엔드 투 엔드 예제는 다음 Notebook을 참조하세요.

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

노트북 받기