Siapkan Stream

Stream mewakili sumber data streaming eksternal, seperti Apache Kafka. Stream menyimpan detail koneksi, autentikasi, skema, dan konfigurasi penyerapan. Setelah streaming dibuat, Anda dapat mereferensikannya menggunakan definisi Tampilan Fitur untuk membuat fitur streaming real time.

Stream memiliki nama tiga bagian (catalog.schema.stream_name). Akses ke Stream diatur oleh tabel penyerapan terkait. Lihat Ingestion dan backfill untuk detailnya.

Persyaratan

Untuk menjalankan perintah notebook: tanpa server atau kluster komputasi klasik yang menjalankan Databricks Runtime 17.0 ML atau lebih tinggi.
feature-engineering-client Paket Python versi 0.18.0 atau lebih tinggi harus diinstal.

Menghubungkan ke sumber streaming

Sebelum menentukan fitur streaming, sambungkan dan uji koneksi alur Lakeflow streaming ke broker Kafka Anda. Feature Store mengandalkan SDP tanpa server, sehingga Anda memerlukan mekanisme untuk menghubungkan komputasi klasik (broker atau endpoint) ke komputasi tanpa server Databricks. Hal ini dilakukan melalui produk seperti privatelink atau dengan memungkinkan komputasi klasik Anda dapat diakses dari internet publik.

Membuat aliran

Gunakan create_stream() untuk membuat Stream baru. Stream memerlukan empat komponen konfigurasi:

Konfigurasi sumber: Menentukan platform streaming dan detail spesifik sumber, seperti langganan topik untuk sumber Kafka.
Konfigurasi koneksi: Menentukan cara menyambungkan dan mengautentikasi ke platform streaming, termasuk server bootstrap dan kredensial.
Konfigurasi skema: Menentukan struktur kunci dan nilai pesan.
Konfigurasi penyerapan: Menentukan di mana dan bagaimana data aliran diserap. Lihat Ingestion dan backfill untuk detailnya.

Untuk pengaturan spesifik source_config sumber dan koneksi, beserta contoh lengkap create_stream() , lihat Apache Kafka. Opsi skema dan penyerapan digunakan bersama di seluruh sumber.

Apache Kafka

Untuk streaming dari Apache Kafka, gunakan KafkaStreamConfig sebagai konfigurasi sumber dan koneksi Unity Catalog untuk autentikasi. Lihat Streaming pada komputasi serverless dan Menyambung ke Apache Kafka untuk konektivitas 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"
        ),
    ),
)

Mode langganan Kafka

Mode langganan menentukan bagaimana Stream memilih topik Kafka yang akan dikonsumsi. Tiga mode didukung:

Modus	Description	Example
`subscribe`	Daftar nama topik yang dipisahkan koma	`KafkaSubscriptionMode(subscribe="topic1,topic2")`
`subscribe_pattern`	Pencocokan pola regex Java untuk nama topik	`KafkaSubscriptionMode(subscribe_pattern="events-.*")`
`assign`	JSON menentukan penetapan partisi topik	`KafkaSubscriptionMode(assign='{"my-topic": [0, 1, 2]}')`

Otentikasi Kafka

Koneksi Katalog Unity (disarankan)

Gunakan koneksi Unity Catalog untuk mengautentikasi ke kluster Kafka Anda. Ini adalah pendekatan yang direkomendasikan untuk autentikasi terkelola. Untuk membuat koneksi, lihat Membuat koneksi. Pembuat Stream harus memiliki USE CONNECTION pada koneksi tersebut. Setiap pengguna yang mematerialisasikan fitur dengan menggunakan Stream sebagai sumber juga harus memiliki USE CONNECTION pada koneksi tersebut.

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

Koneksi mendukung baik IAM (kredensial layanan) maupun autentikasi SASL.

IAM (kredensial layanan)

Autentikasi dengan kredensial layanan Unity Catalog, misalnya untuk terhubung ke Amazon MSK dengan IAM. Untuk membuat kredensial layanan, lihat Membuat kredensial layanan. Atur nama kredensial layanan dengan credential opsi:

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

Selain USE CONNECTION pada koneksi, identitas yang menggunakan kredensial layanan juga harus ada ACCESS di dalamnya. Memberikan ACCESS kredensial layanan yang direferensikan kepada pembuat Stream dan kepada identitas apa pun yang mewujudkan fitur dengan Stream. Lihat Memberikan izin untuk menggunakan kredensial layanan untuk mengakses layanan cloud eksternal.

SASL

Autentikasi SASL menggunakan nama pengguna dan kata sandi. Atur sasl_mechanism ke salah satu hal berikut ini:

PLAIN
SCRAM-SHA-256
SCRAM-SHA-512

Berikan kredensial dengan opsi user dan password. Koneksi menyimpan kredensial ini dengan aman.

Contoh berikut menggunakan SASL/SCRAM. Untuk SASL/PLAIN, atur sasl_mechanism ke 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 langsung

Untuk autentikasi mTLS langsung, berikan file keystore dan truststore yang disimpan pada volume Unity Catalog, dengan kata sandi yang direferensikan melalui cakupan rahasia Databricks. Untuk informasi selengkapnya tentang autentikasi SSL dengan Kafka, lihat Menggunakan SSL untuk menyambungkan Azure Databricks ke Kafka.

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

Konfigurasi skema

Definisikan struktur kunci dan nilai pesan sehingga ingestion dan definisi fitur dapat membaca bidang-bidang individual. Untuk sumber Kafka, payload_schema sesuai dengan nilai pesan Kafka ( value dalam model kunci-nilai Kafka) dan key_schema sesuai dengan kunci pesan Kafka. Setidaknya salah satu dari payload_schema atau key_schema harus disediakan.

Masing-masing SchemaConfig menerima salah satu dari tiga format, sesuai dengan cara sumber menserialkan pesannya: json_schema, , avro_schemaatau proto_schema. Jika tidak ada skema yang disediakan untuk kunci atau payload, itu diperlakukan sebagai string sederhana.

Contoh kode dalam bagian ini menggunakan skema yang dideklarasikan secara sejajar dengan DirectSchemas, di mana skema disediakan sebagai string. Untuk mengelola skema menggunakan registry skema eksternal, lihat Schema registry untuk detailnya.

Skema JSON

Berikan string Skema JSON ke 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"}'
    ),
)

Skema Avro

Berikan string skema Avro ke avro_schema. Tipe logika Avro didukung, termasuk timestamp-millis, date, dan 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"}}'
            '  ]'
            '}'
        )
    ),
)

Skema Protobuf

Berikan ProtoSchemaSpec kepada proto_schema dengan teks sumber Protocol Buffers.proto dan nama pesan payload. Impor ProtoSchemaSpec dari databricks.feature_engineering.entities.

message_name harus berupa nama pesan berkualifikasi lengkap, termasuk package yang dideklarasikan dalam teks .proto (misalnya, com.example.Event, bukan Event). Baik sintaks proto2 maupun proto3 didukung.

google.protobuf.Timestamp dan tipe pembungkus skalar (StringValue, Int32Value, dan seterusnya) didukung, dan impornya diselesaikan secara otomatis. Tipe lain yang sudah dikenal, seperti Duration, Struct, dan Any, ditolak; enkode nilai-nilai tersebut sebagai skalar atau pesan yang didukung sebagai gantinya. Tipe fixed32map skalar dan fixed64 dengan kunci non-string juga tidak didukung.

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

Mendekode data menggunakan skema

Databricks mendekodekan setiap pesan dengan fungsi from_avro, from_protobuf, dan from_json milik Spark. Perilaku berikut berlaku baik Anda mendeklarasikan skema secara inline atau menyelesaikannya dari registri skema:

Rekaman tidak valid. Pendekodean menggunakan mode PERMISSIVE, sehingga record yang tidak sesuai dengan skemanya akan menghasilkan nilai null alih-alih menyebabkan aliran gagal.
Serikat Avro. Gabungan beberapa tipe rekaman didekodekan menjadi struktur dengan satu bidang untuk setiap tipe rekaman, yang masing-masing dinamai berdasarkan rekaman Avro-nya.
Tipe Protobuf. Bilangan bulat tanpa tanda didekodekan ke tipe bertanda yang lebih luas (misalnya, uint32 ke uint64 dan BIGINT ke DECIMAL(20,0)), field enum didekodekan ke nama stringnya, dan tipe pembungkus skalar (misalnya, StringValue dan Int32Value) didekodekan ke kolom nullable dari tipe yang dibungkus.

Registri skema

Registri skema menyimpan dan skema versi yang digunakan produser dan konsumen streaming, menegakkan aturan kompatibilitas seiring perkembangan skema tersebut. Ketika registry skema eksternal dikonfigurasi, Feature Store membaca skema dari registry dan menggunakannya untuk mendekode pesan streaming. Anda tidak mendeklarasikan skema secara inline pada Stream saat menggunakan registri skema.

Dukungan registri skema memiliki keterbatasan berikut:

Hanya didukung untuk stream Kafka.
Hanya Confluent Schema Registry yang didukung
Hanya format Avro dan Protobuf yang didukung. Untuk membaca pesan JSON, deklarasikan skema secara inline sebagai gantinya. Lihat skema JSON.
Setiap Stream terhubung tepat ke satu subjek Confluent untuk nilai pesan, dan satu untuk kunci pesan (jika disediakan). Topik stream yang berisi beberapa catatan skema bukanlah konfigurasi yang didukung. Jika Stream Anda terhubung ke topik yang berisi beberapa skema, catatan yang tidak sesuai dengan skema untuk subjek yang ditentukan akan didekode sebagai null.

Hubungkan ke registri skema

Berikan detail koneksi registry sebagai opsi pada koneksi Kafka Unity Catalog, dan simpan rahasia API registry dalam lingkup rahasia Databricks. Identitas run-as Stream harus memiliki READ izin pada lingkup rahasia, karena pipeline ingestion membaca rahasia tersebut saat runtime. Untuk cara membuat dan mengonfigurasi koneksi, lihat Buat koneksi.

Tambahkan schema_registry_urlopsi , schema_registry_api_key, dan schema_registry_api_secret ke koneksi yang digunakan untuk autentikasi. Contoh berikut membuat koneksi Kafka yang mengautentikasi ke broker dengan kredensial layanan Unity Catalog dan ke registry dengan kunci API:

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

Atur opsi schema_registry_api_secret pada koneksi Kafka dan referensi cakupan rahasia pada Stream ke rahasia yang sama.

Buat aliran yang menggunakan registri skema

Teruskan SchemaRegistryConfig sebagai schema_config. Referensikan rahasia API registry dengan api_secret_ref, dan identifikasi subjek serta format dengan payload_schema_locator untuk nilai pesan, atau key_schema_locator untuk kunci pesan. Setidaknya satu pelacak harus disediakan.

Perhatikan perbedaan di sini dibandingkan dengan contoh skema langsung pada bagian konfigurasi Skema . Saat menggunakan registri skema, Anda tidak menyediakan skema secara langsung di Stream ke schema_config. Sebagai gantinya, Anda menentukan a SchemaRegistryConfig yang mengidentifikasi skema di registry.

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

Subjek Confluent adalah lingkup bernama di mana riwayat versi skema didaftarkan dan kompatibilitas ditegakkan. Setel subject ke nama cakupan yang relevan, yang umumnya ditentukan berdasarkan strategi penamaan subjek:

TopicNameStrategy (default, menurunkan subjek dari nama topik): <topic>-value untuk nilai dan <topic>-key untuk kunci. Misalnya, skema nilai untuk topik transactions menggunakan subjek transactions-value.
RecordNameStrategy (menentukan subjek dari nama rekaman skema, tanpa bergantung pada topik): nama rekaman yang sepenuhnya memenuhi syarat, seperti com.example.Payment. Ini adalah namespace dan nama record untuk Avro, atau paket dan nama pesan untuk Protobuf.
TopicRecordNameStrategy (menggabungkan nama topik dan catatan): <topic>-<fully-qualified-record-name>, seperti transactions-com.example.Payment.

format diperlukan. Atur ke SchemaLocatorFormat.FORMAT_AVRO atau SchemaLocatorFormat.FORMAT_PROTOBUF agar sesuai dengan cara topik diserialisasikan.

Perubahan skema

Pipeline penyerapan data menentukan skema subjek saat ini ketika dimulai. Ketika Anda mendaftarkan versi skema kompatibel mundur baru pada subjek di registri skema, pipeline yang berjalan akan tetap menggunakan versi awalnya.

Untuk Stream yang didukung registri skema, pipeline ingestion dimulai ulang secara otomatis setiap beberapa jam. Setiap kali memulai ulang, ia akan mengambil versi skema terbaru subjek, dan kolom baru atau yang diubah muncul di tabel ingestion.

Stream yang menggunakan skema langsung alih-alih registri skema melakukan evolusi skema dengan update_stream. Lihat Perbarui stream.

Untuk bagaimana pipeline menangani catatan yang tidak sesuai dengan skema yang sedang digunakan, lihat Decoding data using schemas.

Saring rekaman berdasarkan jenis

Stream mendekode setiap catatan menggunakan satu skema kunci dan nilai (jika disediakan), baik Anda menentukannya langsung maupun menggunakan registri skema. Karena sebuah topik dapat memuat lebih dari satu jenis catatan dan Stream dapat berlangganan beberapa topik, gunakan record_type_filter untuk memilih catatan dari topik tersebut mana yang termasuk dalam Stream ini.

Sediakan ekspresi SQL yang merujuk pada bidang yang telah didekode dengan notasi titik, misalnya value.event_type = 'transaction'. Rekaman yang tidak sesuai dengan filter akan diabaikan. Mereka tidak ditulis ke tabel ingestion dan tidak digunakan dalam materialisasi. Untuk membuat Stream untuk tipe rekaman lain, buat Stream terpisah dengan record_type_filter yang berbeda.

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

Bahkan tanpa record_type_filter, proses dekode tidak pernah menyebabkan Stream gagal. Rekaman yang tidak sesuai dengan skema yang dikonfigurasi akan didekode secara permisif. Rekaman tersebut didekodekan dengan salah satu cara berikut:

Menjadi baris dengan nilai NULL untuk kolom yang diharapkan oleh skema tetapi tidak ada dalam record (JSON, Avro, dan Protobuf).
Ke baris yang menampung nilai milik jenis rekaman yang berbeda (khusus Avro dan Protobuf).

Untuk mengidentifikasi apakah baris dalam tabel ingestion termasuk dalam tipe rekaman yang diharapkan, gunakan salah satu pemeriksaan berikut:

Periksa bahwa sebuah bidang sama dengan nilai harapan, misalnya value.event_type = 'transaction' (lebih diutamakan untuk Avro dan Protobuf).
Periksa bahwa sebuah bidang bukan NULL kosong, misalnya value.activity_id IS NOT NULL.

Penggunaan record_type_filter dengan Stream terpisah disarankan ketika skema berbeda secara signifikan antara tipe rekaman pada topik tersebut atau Anda ingin mengatur akses ke setiap tipe rekaman secara independen. Agar biaya tetap terkendali, Databricks merekomendasikan agar Anda mempertahankan jumlah Stream tetap sedikit, karena setiap Stream memiliki pipeline penyerapan data sendiri dan tabel penyerapan terpisah. Setiap Stream juga menggunakan sumber daya komputasi yang terpisah saat materialisasi. Anda dapat menggunakan filter khusus fitur untuk materialisasi.

record_type_filter berbeda dari filter_condition suatu fitur. record_type_filter diatur pada Stream dan mengontrol rekaman mana yang dimasukkan dan tersedia untuk semua fitur menggunakan Stream sebagai sumber, sedangkan filter_condition diatur pada fitur individual dan memfilter baris sebelum agregasi. Lihat Kondisi filter pada sumber streaming untuk detail lebih lanjut tentang filter_condition.

Konsumsi dan pengisian kembali

Parameter ingestion_config mengonfigurasi bagaimana data aliran diambil dan disimpan untuk pelatihan dan penyajian.

Akses ke Stream diatur oleh tabel penyerapan:

SELECT pada tabel ingestion memberikan akses baca ke Stream.
MANAGE pada tabel penyerapan memberikan akses penghapusan.

Untuk informasi selengkapnya tentang hak istimewa tabel, lihat Tabel dan Referensi hak istimewa Unity Catalog.

Alur penyerapan

Ketika sebuah aliran dibuat, Databricks memulai pipeline penyerapan data terkelola yang terus-menerus membaca pesan dari aliran sumber dan menuliskannya ke dalam tabel Delta (tabel penyerapan). Pipeline dimulai dari posisi terbaru di sumber dan berjalan terus-menerus, hanya menangkap pesan baru yang tiba setelah stream dibuat. Tabel penyerapan ini digunakan untuk pelatihan dengan fitur streaming. Saat aliran dihapus, alur penyerapan dan tabel penyerapannya juga dihapus.

Tujuan penyerapan

ingestion_destination menentukan nama tabel Delta tiga bagian tempat data aliran ditulis.

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

Skema tabel penyerapan

Tabel ingestion berisi data pesan beserta kolom metadata. Kolom umum tersedia untuk setiap sumber; kolom kafka_* hanya tersedia untuk stream Kafka.

Column	Tipe	Source	Description
`key`	Berbeda-beda (dari `key_schema`)	Common	Kunci pesan, disusun sesuai dengan skema yang Anda berikan.
`value`	Berbeda-beda (dari `payload_schema`)	Common	Nilai pesan (payload), disusun sesuai dengan skema yang Anda berikan.
`stream_record_timestamp`	`TIMESTAMP`	Common	Tanda waktu rekaman. Untuk data forward-fill, ini adalah stempel waktu ingest sumber. Untuk data backfill, data ini disediakan oleh pelanggan.
`record_source`	`STRING`	Common	Baik `"stream"` (forward-fill dari live stream) atau `"backfill"` (dari sumber backfill).
`kafka_topic`	`STRING`	Kafka	Topik Kafka tempat rekaman dikonsumsi.
`kafka_partition`	`INT`	Kafka	Partisi Kafka tempat rekaman dikonsumsi.
`kafka_offset`	`LONG`	Kafka	Offset Kafka dari rekaman dalam partisinya.

Sumber isi ulang

Karena pipeline forward-fill dimulai dari posisi terbaru pada sumber, pipeline ini tidak menangkap pesan yang sudah ada sebelum stream tersebut dibuat. Untuk memberikan cakupan data historis untuk pelatihan, konfigurasikan sumber isi ulang opsional.

Saat sumber backfill dikonfigurasi, Databricks menjalankan job satu kali MERGE INTO yang menyalin baris backfill ke tabel ingesti dengan record_source="backfill". MERGE berjalan hanya setelah pemeriksa tumpang tindih mengonfirmasi bahwa sumber isi ulang dan aliran pengisian maju memiliki tanda waktu yang tumpang tindih (lihat Tumpang tindih antara isi ulang dan data streaming langsung). Jika kondisi tumpang tindih tidak terpenuhi dalam 2 hari, MERGE tetap berjalan untuk menghindari pemblokiran tanpa batas waktu.

Tabel backfill harus memiliki kolom stream_record_timestamp bertipe TIMESTAMP dalam zona waktu UTC. Kolom metadata lainnya akan tetap diteruskan jika ada pada sumber backfill, atau diatur ke NULL jika tidak. Untuk Kafka, ini adalah kafka_topic, kafka_partition, dan kafka_offset.

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

Tumpang tindih antara data isi ulang dan streaming langsung

Sebelum menjalankan MERGE antara isi ulang dan tabel penyerapan, pemeriksaan tumpang tindih membandingkan tanda waktu pada dua tabel:

Maks isi ulang: Maksimum stream_record_timestamp di sumber isi ulang.
Penyerapan min: Minimum stream_record_timestamp baris (record_source="stream") dalam tabel penyerapan.

MERGE berlanjut ketika tanda waktu terbaru isi ulang melebihi tanda waktu tercepat tabel penyerapan setidaknya 1 jam. Tumpang tindih ini memastikan tidak ada celah dalam tabel penyerapan. Jika kondisi tumpang tindih tidak terpenuhi dalam 2 hari, MERGE tetap berjalan untuk menghindari pemblokiran tanpa batas waktu.

Karena pipeline penyerapan data dimulai dari posisi terbaru pada sumber, pipeline ini hanya menangkap pesan yang masuk setelah stream dibuat. Sumber isi ulang Anda harus berisi data yang meluas ke rentang waktu penyerapan — bukan hanya hingga waktu pembuatan aliran.

Misalnya, jika Anda membuat streaming pada pukul 15.00, alur forward-fill mulai membaca pesan dari pukul 15.00 dan seterusnya. Sumber isi ulang Anda harus menyertakan data dengan tanda waktu hingga setidaknya pukul 16.00 (1 jam melewati awal pengisian maju) untuk memenuhi pemeriksaan tumpang tindih. Ini berarti Anda harus memperbarui tabel isi ulang setelah pukul 16.00 untuk memastikan tabel penyerapan tidak memiliki celah.

Deduplication

Gunakan deduplication_columns untuk menentukan path kolom guna mengidentifikasi baris duplikat selama proses ingesti antara data stream backfill dan forward-fill. Gunakan notasi titik untuk bidang berlapis (misalnya, "value.user_id").

Pilih kolom deduplikasi berdasarkan data Anda:

Jika setiap rekaman di aliran Anda berisi pengidentifikasi unik (misalnya, value.transaction_id), gunakan kolom tersebut untuk deduplikasi.
Jika sumber backfill Anda menyertakan kolom kafka_partition dan kafka_offset, gunakan kolom tersebut untuk mengidentifikasi setiap catatan secara unik.
Jika tidak ada kolom deduplikasi yang ditentukan, kunci deduplikasi default adalah kombinasi lengkap dari key, , valuedan stream_record_timestamp. Ini tidak disarankan karena pencocokan kriteria ketat ini dapat dengan mudah menyebabkan duplikat.

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

Atribusi biaya

Atur tags dan budget_policy_id pada IngestionConfig untuk mengatribusikan biaya penyerapan terkelola Stream. Azure Databricks menerapkannya pada pipeline ingestion Lakeflow dan pekerjaan forward-fill serta backfill saat Stream dibuat.

Sebagai contoh, untuk melihat batasan tag dan cara mengkueri pengeluaran yang diatribusikan, lihat Mengatribusikan biaya dengan tag dan kebijakan penggunaan serverless.

Keluarkan kolom dari aliran

Gunakan excluded_columns untuk menghapus kolom tertentu dari Stream yang tidak ingin Anda ingest. Kolom yang dikecualikan tidak ditulis ke tabel ingestion dan tidak dapat direferensikan oleh fitur atau digunakan dalam pelatihan.

Tentukan setiap kolom menggunakan notasi titik dalam kunci atau nilai pesan, seperti value.user.email atau key.account_id. Kolom-kolom ini dihapus dari hasil dekode key dan value di seluruh proses ingesti, backfill, dan materialisasi. Jika suatu jalur menunjuk ke sebuah struct, semua field bersarang di dalamnya juga akan dihapus (misalnya, menghapus value.address juga akan menghapus value.address.city dan 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"],
)

Saat menggunakan skema langsung, kolom yang dikecualikan harus sudah ada di dalam skema kunci atau nilai, atau create_stream akan gagal. Saat menggunakan registri skema, Anda dapat mengecualikan kolom sebelum kolom itu ada. Kolom yang dikecualikan juga tidak dapat menjadi kolom deduplikasi, karena kolom deduplikasi diperlukan untuk mengidentifikasi baris duplikat. Setiap fitur yang merujuk pada kolom yang dikecualikan (misalnya, sebagai entitas, deret waktu, atau input) gagal dibuat.

Anda dapat mengubah kolom yang dikecualikan dari Stream setelah dibuat dengan update_stream, baik pada Stream yang didukung oleh skema langsung maupun Stream yang didukung schema registry. Lihat Update a stream untuk informasi selengkapnya.

Kelola aliran

Mendapatkan aliran

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

Mencantumkan aliran

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

Atur include_schemas=True untuk menyertakan detail skema lengkap. Skema bisa besar dan ini mungkin mengakibatkan operasi yang berjalan lama. Untuk mengambil skema satu per satu, gunakan get_stream.

Perbarui stream

Gunakan update_stream untuk mengubah Stream setelah pembuatan. Berikan schema_config untuk mengembangkan skema langsung, excluded_columns untuk mengubah kolom mana yang dihapus, atau keduanya. Memperbarui bidang lain tidak didukung. Buat Stream baru sebagai gantinya.

Memperbarui Stream memulai kembali pipeline intake-nya sehingga perubahan tersebut berlaku. Proses ingest biasanya dilanjutkan dalam beberapa menit.

Ubah skema langsung

Untuk Stream yang menggunakan skema langsung, berikan a DirectSchemas ke schema_config. Set payload_schema, key_schema, atau keduanya. Sisi yang tidak kamu atur tetap tidak berubah. Stream yang didukung oleh registri skema menolak pembaruan schema_config dan sebagai gantinya harus dievolusikan melalui registri.

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

Pembaruan skema harus kompatibel dengan versi sebelumnya agar pipeline ingestion yang sedang berjalan dapat terus mendekode rekaman yang sudah ada dan menulis ke tabel ingestion. Perubahan lain ditolak.

Apa yang diizinkan tergantung pada formatnya:

JSON dan Protobuf: menambahkan bidang opsional, menghapus bidang, dan memperluas tipe bidang (misalnya, int menjadi bigint). Protobuf juga memungkinkan penyusunan ulang kolom.
Avro: hanya mengizinkan pelebaran int menjadi long dan penghapusan field di bagian akhir yang byte-nya tidak dibaca oleh field setelahnya. Untuk mengubah skema Avro dengan lebih bebas, gunakan stream yang didukung registry skema sebagai gantinya.

Menambahkan field akan memperbesar struct key dan value yang didekode dari tabel ingestion. Baris yang ditulis sebelum pembaruan tetap mempertahankan bentuk aslinya, dan field yang ditambahkan akan terbaca sebagai NULL pada baris-baris sebelumnya. Penghapusan dan perubahan tipe hanya berlaku untuk rekaman yang dimasukkan setelah pembaruan.

Ubah kolom yang dikecualikan

Teruskan kumpulan lengkap jalur kolom baru ke excluded_columns, yang menggantikan kumpulan yang ada. Berikan daftar kosong ([]) untuk menghapus semua pengecualian. Untuk detail selengkapnya tentang perilaku ini, lihat Mengecualikan kolom dari aliran data.

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

Perubahan pada kolom yang dikecualikan hanya dapat dilakukan ke depan. Kolom yang baru dikecualikan berhenti ditulis (muncul di NULL) dan kolom yang baru disertakan mulai terisi ke depannya, sementara baris yang sebelumnya ditulis dibiarkan as-is. Untuk mencegah kolom baru pernah diimpor:

Schema registry: tambahkan kolom ke excluded_columns terlebih dahulu dan tunggu pipeline ingestion dimulai ulang, lalu daftarkan versi skema baru di registry.
Skema langsung: tambahkan kolom ke schema_config dan ke excluded_columns dalam pemanggilan update_stream yang sama.

Menghapus aliran

Menghapus aliran juga menghapus alur penyerapan dan tabel penyerapannya.

Warning

Model atau fitur apa pun yang mereferensikan aliran yang dihapus tidak akan lagi memiliki akses ke data aliran yang mendasar. Buat salinan tabel penyerapan sebelum penghapusan jika Anda memerlukan data ini tetapi tidak lagi memerlukan aliran.

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

Contoh buku catatan

Untuk contoh menyeluruh yang membuat Stream, mendefinisikan fitur streaming, dan menerapkan ke endpoint penyajian, lihat notebook berikut:

Buku catatan mulai cepat Tampilan Fitur Streaming

Dapatkan buku catatan

Saran dan Komentar

Apakah halaman ini membantu?

Terakhir diperbarui pada 2026-10-12