Peristiwa dan feed

Datastream mengatur data yang direplikasi ke dalam hierarki tiga tingkat yang terdiri dari aliran data, objek, dan peristiwa. Aliran data menghubungkan sumber data ke tujuan, sedangkan objek mewakili entity sumber tertentu dalam aliran data tersebut, seperti tabel database. Peristiwa menangkap satu perubahan data dari objek, seperti penyisipan database.

Setiap peristiwa menyertakan data dan metadata yang mengidentifikasi aliran data dan objek sumbernya. Anda dapat menggunakan metadata ini untuk melacak asal perubahan, menghapus data duplikat di tujuan, dan mempertahankan pengurutan di seluruh konsumen downstream.

Tentang peristiwa

Setiap peristiwa terdiri dari tiga jenis data:

• Data peristiwa: Bagian ini menunjukkan perubahan pada data itu sendiri dari objek yang berasal dari sumber aliran data. Setiap peristiwa berisi keseluruhan baris yang diubah.
• Metadata umum: Metadata ini muncul di setiap peristiwa yang dihasilkan oleh Datastream yang digunakan untuk tindakan, seperti menghapus data duplikat di tujuan.
• Metadata yang spesifik per sumber: Metadata ini muncul di setiap peristiwa yang dihasilkan oleh sumber streaming tertentu. Metadata ini bervariasi menurut sumber.

Data peristiwa

Data peristiwa adalah payload dari setiap perubahan dari objek tertentu yang berasal dari sumber aliran data.

Peristiwa menggunakan format Avro atau JSON.

Saat menggunakan format Avro, untuk setiap kolom, peristiwa berisi indeks dan nilai kolom. Dengan menggunakan indeks kolom, nama kolom dan jenis terpadu dapat diambil dari skema di header Avro.

Catatan: Saat Datastream mengonversi data ke format Avro, karakter khusus dalam nama kolom akan diganti dengan garis bawah. Hal ini dapat menyebabkan nama kolom duplikat dan mengakibatkan aliran data Anda gagal. Untuk mengetahui informasi selengkapnya tentang konvensi penamaan di Avro, lihat dokumentasi Apache Avro.

Saat menggunakan format JSON, untuk setiap kolom, peristiwa berisi nama dan nilai kolom.

Metadata peristiwa dapat digunakan untuk mengumpulkan informasi tentang asal peristiwa, serta untuk menghapus data duplikat di tujuan dan mengurutkan peristiwa oleh konsumen downstream.

Tabel berikut mencantumkan dan menjelaskan kolom serta jenis data untuk metadata peristiwa umum dan khusus sumber.

Metadata umum

Metadata ini konsisten di seluruh aliran data dari semua jenis.

Kolom	Jenis Avro	Jenis JSON	Deskripsi
stream_name	string	string	Nama unik streaming seperti yang ditentukan pada waktu pembuatan.
read_method	string	string	Menunjukkan apakah data dibaca dari sumber menggunakan metode change data capture (CDC), sebagai bagian dari pengisian ulang historis, atau sebagai bagian dari tugas tambahan yang dibuat saat transaksi di-roll back selama replikasi CDC. Nilai yang memungkinkan mencakup: cdc backfill oracle-cdc-logminer oracle-backfill oracle-supplementation mysql-cdc-binlog mysql-backfill-incremental mysql-backfill-fulldump postgres-cdc-wal postgresql-backfill sqlserver-cdc sqlserver-backfill salesforce-cdc salesforce-backfill
object	string	string	Nama yang digunakan untuk mengelompokkan berbagai jenis peristiwa, biasanya nama tabel atau objek di sumber.
schema_key	string	string	ID unik untuk skema terpadu peristiwa.
uuid	string	string	ID unik untuk peristiwa yang dihasilkan oleh Datastream.
read_timestamp	timestamp-millis	string	Stempel waktu (UTC) saat data dibaca oleh Datastream (stempel waktu epoch dalam milidetik).
source_timestamp	timestamp-millis	string	Stempel waktu (UTC) saat data berubah di sumber (stempel waktu epoch dalam milidetik).
sort_keys	{"type": "array", "items": ["string", "long"]}	array	Array nilai yang dapat digunakan untuk mengurutkan peristiwa sesuai urutan terjadinya.

Metadata yang spesifik per sumber

Metadata ini terkait dengan peristiwa CDC dan pengisian ulang dari database sumber. Untuk melihat metadata ini, pilih sumber dari menu drop-down berikut.

Semua sumber MySQL Oracle PostgreSQL SQL Server Spanner Salesforce MongoDB

Sumber	Kolom	Jenis Avro	Jenis JSON	Deskripsi
MySQL	log_file	string	string	File log tempat Datastream mengambil peristiwa dalam replikasi CDC saat menggunakan replikasi berbasis binlog.
MySQL	log_position	long	long	Posisi log (offset) dalam log biner MySQL saat menggunakan replikasi berbasis binlog.
MySQL	binlog_file	string	string	File log biner tempat Datastream mengambil peristiwa dalam replikasi CDC saat menggunakan replikasi berbasis GTID.
MySQL	binlog_position	long	long	Posisi (offset) dalam log biner MySQL saat menggunakan replikasi berbasis GTID.
MySQL	gtid	string	string	ID transaksi global (GTID) yang secara unik mengidentifikasi transaksi untuk peristiwa saat menggunakan replikasi berbasis GTID.
MySQL	datastream_master_server_uuid	string	string	UUID instance server sumber MySQL tempat Datastream membaca peristiwa saat menggunakan replikasi berbasis GTID.
MySQL	datastream_master_server_id	long	long	ID server instance server sumber MySQL tempat Datastream membaca peristiwa saat menggunakan replikasi berbasis GTID.
MySQL	primary_keys	array string	array string	Daftar (satu atau beberapa) nama kolom yang membentuk kunci utama tabel. Jika tabel tidak memiliki kunci utama, kolom ini akan kosong.
MySQL	is_deleted	boolean	boolean	Nilai true menunjukkan bahwa baris telah dihapus di sumber. Nilai false menandakan bahwa baris tidak dihapus.
MySQL	database	string	string	Database yang terkait dengan peristiwa.
MySQL	table	string	string	Tabel yang terkait dengan peristiwa.
MySQL	change_type	string	string	Jenis perubahan (INSERT, UPDATE-INSERT, UPDATE-DELETE, dan DELETE) yang diwakili oleh peristiwa. Untuk replikasi berbasis baris MySQL, setiap peristiwa untuk baris dengan perubahan kunci utama berisi dua image: image "sebelum" yang kolomnya dicocokkan saat menelusuri baris yang akan diperbarui, dan image "setelah" yang berisi perubahan. Misalnya, untuk perubahan UPDATE-DELETE, UPDATE-DELETE adalah gambar "sebelum" dan UPDATE-INSERT adalah gambar "setelah". Pelajari lebih lanjut replikasi berbasis baris MySQL.
Oracle	log_file	string	string	File log tempat Datastream mengambil peristiwa dalam replikasi CDC.
Oracle	scn	long	long	Posisi log (offset) dalam log transaksi Oracle.
Oracle	row_id	string	string	row_id Oracle.
Oracle	is_deleted	boolean	boolean	Nilai true menunjukkan bahwa baris telah dihapus di sumber. Nilai false menandakan bahwa baris tidak dihapus.
Oracle	database	string	string	Database yang terkait dengan peristiwa.
Oracle	schema	string	string	Skema yang terkait dengan tabel dari peristiwa.
Oracle	table	string	string	Tabel yang terkait dengan peristiwa.
Oracle	change_type	string	string	Jenis perubahan (INSERT, UPDATE-INSERT, UPDATE-DELETE, dan DELETE) yang diwakili oleh peristiwa. Untuk replikasi berbasis baris Oracle, setiap peristiwa untuk baris dengan perubahan kunci utama berisi dua image: image "sebelum" yang kolomnya dicocokkan saat menelusuri baris yang akan diupdate, dan image "setelah" yang berisi perubahan. Misalnya, untuk perubahan UPDATE-DELETE, UPDATE-DELETE adalah gambar "sebelum" dan UPDATE-INSERT adalah gambar "setelah". Pelajari lebih lanjut replikasi berbasis baris Oracle.
Oracle	tx_id	string	string	ID transaksi yang terkait dengan peristiwa.
Oracle	rs_id	string	string	ID set data. Penggabungan rs_id dan ssn secara unik mengidentifikasi baris di V$LOGMNR_CONTENTS. rs_id mengidentifikasi secara unik data pengulangan yang menghasilkan baris.
Oracle	ssn	long	long	Nomor urut SQL. Nomor ini digunakan dengan rs_id dan secara unik mengidentifikasi baris di V$LOGMNR_CONTENTS.
PostgreSQL	schema	string	string	Skema yang terkait dengan tabel dari peristiwa.
PostgreSQL	table	string	string	Tabel yang terkait dengan peristiwa.
PostgreSQL	is_deleted	boolean	boolean	Nilai true menunjukkan bahwa baris telah dihapus di sumber. Nilai false menandakan bahwa baris tidak dihapus.
PostgreSQL	change_type	string	string	Jenis perubahan (INSERT, UPDATE, DELETE) yang diwakili oleh peristiwa. Untuk replikasi berbasis baris PostgreSQL, setiap peristiwa untuk baris dengan perubahan kunci utama berisi dua image: image "sebelum" yang kolomnya dicocokkan saat menelusuri baris yang akan diperbarui, dan image "setelah" yang berisi perubahan. Misalnya, untuk perubahan UPDATE-DELETE, UPDATE-DELETE adalah gambar "sebelum" dan UPDATE-INSERT adalah gambar "setelah". Pelajari lebih lanjut replikasi berbasis baris PostgreSQL.
PostgreSQL	tx_id	string	string	ID transaksi yang terkait dengan peristiwa.
PostgreSQL	lsn	string	string	Nomor urut log untuk entri saat ini.
PostgreSQL	primary_keys	array string	array string	Daftar (satu atau beberapa) nama kolom yang membentuk kunci utama tabel. Jika tabel tidak memiliki kunci utama, kolom ini akan kosong.
SQL Server	table	string	string	Tabel yang terkait dengan peristiwa.
SQL Server	database	long	long	Database yang terkait dengan peristiwa.
SQL Server	schema	array string	array string	Skema yang terkait dengan tabel dari peristiwa.
SQL Server	is_deleted	boolean	boolean	Nilai true menunjukkan bahwa baris telah dihapus di sumber. Nilai false menandakan bahwa baris tidak dihapus.
SQL Server	lsn	string	string	Nomor urut log untuk peristiwa.
SQL Server	tx_id	string	string	ID transaksi yang terkait dengan peristiwa.
SQL Server	physical_location	array bilangan bulat	array bilangan bulat	Lokasi fisik data log yang dijelaskan oleh tiga bilangan bulat: ID file, ID halaman, dan ID slot data.
SQL Server	replication_index	array string	array string	Daftar nama kolom indeks yang dapat mengidentifikasi baris dalam tabel secara unik.
SQL Server	change_type	string	string	Jenis perubahan (INSERT, UPDATE, DELETE) yang diwakili oleh peristiwa.
Spanner	commit_timestamp	timestamp-micros	string	Stempel waktu commit peristiwa.
Spanner	snapshot	boolean	boolean	Apakah peristiwa tersebut merupakan peristiwa snapshot pengisian ulang.
Spanner	project_id	string	string	ID project Spanner.
Spanner	instance_id	string	string	ID instance Spanner.
Spanner	database_id	string	string	ID database Spanner.
Spanner	schema	string	string	Skema yang terkait dengan tabel dari peristiwa. Kosong jika tabel tidak berada dalam skema bernama.
Spanner	table	string	string	Tabel Spanner.
Spanner	change_stream_name	string	string	Aliran perubahan Spanner.
Spanner	server_transaction_id	string	string	ID transaksi yang terkait dengan peristiwa.
Spanner	record_sequence	long atau string	long atau string	Urutan data yang terkait dengan peristiwa. Untuk aliran data perubahan yang dibuat dengan mode partisi IMMUTABLE_KEY_RANGE, nilainya adalah nomor urut numerik dari jenis long. Untuk aliran data perubahan yang dibuat dengan mode partisi MUTABLE_KEY_RANGE, nilainya adalah string record_sequence dari data perubahan data, seperti 1cac8f3f38eeaf53-00000000.
Spanner	mod_number	long	long	Nomor modifikasi yang terkait dengan peristiwa.
Spanner	transaction_tag	string	string	Tag untuk transaksi terkait.
Spanner	number_of_records_in_transaction	long	long	Jumlah data dalam transaksi terkait.
Spanner	value_capture_type	string	string	Jenis pengambilan nilai aliran data perubahan. Nilai yang mungkin adalah NEW_ROW dan NEW_ROW_AND_OLD_VALUES.
Spanner	change_type	string	string	Jenis perubahan (INSERT, UPDATE, atau DELETE) yang diwakili oleh peristiwa.
Spanner	primary_keys	array string	array string	Daftar kolom kunci utama untuk tabel.
Spanner	is_deleted	boolean	boolean	Apakah perubahan tersebut mewakili peristiwa DELETE.
Salesforce	object_name	string	string	Nama objek Salesforce yang terkait dengan peristiwa.
Salesforce	domain	string	string	Nama domain yang terkait dengan peristiwa.
Salesforce	is_deleted	boolean	boolean	Nilai true menunjukkan bahwa baris telah dihapus di sumber. Nilai false menandakan bahwa baris tidak dihapus.
Salesforce	change_type	string	string	Jenis perubahan (INSERT, UPDATE, DELETE) yang diwakili oleh peristiwa.
Salesforce	primary_keys	array string	array string	Daftar nama kolom yang membentuk kunci utama tabel. Jika tabel tidak memiliki kunci utama, kolom ini akan kosong.
MongoDB	database	string	string	Database yang terkait dengan peristiwa.
MongoDB	collection	string	string	Koleksi yang terkait dengan peristiwa. Koleksi serupa dengan tabel dalam database relasional.
MongoDB	change_type	string	string	Jenis perubahan (CREATE, UPDATE, dan DELETE) yang diwakili oleh peristiwa.
MongoDB	is_deleted	boolean	boolean	Nilai true menunjukkan bahwa baris telah dihapus di sumber. Nilai false menandakan bahwa baris tidak dihapus.
MongoDB	primary_keys	array string	array string	Kolom _id yang berfungsi sebagai kunci utama untuk setiap dokumen dalam koleksi.

Contoh alur peristiwa

Alur ini mengilustrasikan peristiwa yang dihasilkan oleh tiga operasi berturut-turut: INSERT, UPDATE, dan DELETE, pada satu baris dalam tabel SAMPLE untuk database sumber.

Catatan: Contoh alur peristiwa ini ditujukan untuk database Oracle, tetapi serupa untuk database MySQL atau PostgreSQL.

WAKTU	THIS_IS_MY_PK (int)	FIELD1 (nchar nullable)	FIELD2 (nchar non-null)>
0	1231535353	foo	TLV
1	1231535353	NULL	TLV

INSERT (T0)

Payload pesan terdiri dari keseluruhan baris baru.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "d7989206-380f-0e81-8056-240501101100",
  "read_timestamp": "2019-11-07T07:37:16.808Z",
  "source_timestamp": "2019-11-07T02:15:39",  
  "source_metadata": {
    "log_file": ""
    "scn": 15869116216871,
    "row_id": "AAAPwRAALAAMzMBABD",
    "is_deleted": false,
    "database": "DB1",
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "INSERT",
    "tx_id": 
    "rs_id": "0x0073c9.000a4e4c.01d0",
    "ssn": 67,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": "foo",
    "FIELD2": "TLV",
  }
}

UPDATE (T1)

Payload pesan terdiri dari keseluruhan baris baru. Nilai ini tidak mencakup nilai sebelumnya.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "e6067366-1efc-0a10-a084-0d8701101101",
  "read_timestamp": "2019-11-07T07:37:18.808Z",
  "source_timestamp": "2019-11-07T02:17:39",  
  "source_metadata": {
    "log_file": 
    "scn": 15869150473224,
    "row_id": "AAAGYPAATAAPIC5AAB",
    "is_deleted": false,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "UPDATE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0010",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

DELETE (T2)

Payload pesan terdiri dari keseluruhan baris baru.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "c504f4bc-0ffc-4a1a-84df-6aba382fa651",
  "read_timestamp": "2019-11-07T07:37:20.808Z",
  "source_timestamp": "2019-11-07T02:19:39",
  "source_metadata": {
    "log_file": 
    "scn": 158691504732555,
    "row_id": "AAAGYPAATAAPIC5AAC",
    "is_deleted": true,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "DELETE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0011",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

Pengurutan dan konsistensi

Bagian ini membahas cara Datastream menangani pengurutan dan konsistensi.

Memesan

Datastream tidak menjamin pengiriman peristiwa secara berurutan, tetapi setiap peristiwa berisi baris data lengkap, stempel waktu saat data ditulis ke sumber, dan metadata pengurutan.

Peristiwa yang merupakan bagian dari pengisian ulang awal memiliki kolom read_method yang ditetapkan ke backfill atau diakhiri dengan -backfill, -fulldump, atau -incremental. Peristiwa pengisian ulang dibuat tidak berurutan sesuai desain dan dapat digunakan dalam urutan apa pun.

Peristiwa yang merupakan bagian dari replikasi change data capture (CDC) yang sedang berlangsung memiliki kolom read_method yang ditetapkan ke cdc atau nilai CDC khusus sumber, seperti mysql-cdc-binlog, oracle-cdc-logminer, postgres-cdc-wal, sqlserver-cdc, atau salesforce-cdc. Cara Anda mengurutkan peristiwa replikasi berkelanjutan di semua sumber bergantung pada tujuan Anda:

Tujuan	Deskripsi
BigQuery	Cara pengurutan peristiwa di BigQuery bergantung pada mode tulis yang Anda konfigurasi untuk aliran data: Mode tulis penggabungan: BigQuery menggabungkan peristiwa yang tidak berurutan dalam urutan yang benar secara otomatis menggunakan metadata peristiwa dan nomor urutan perubahan (CSN) internal. Tabel tanpa kunci utama ditulis dalam mode hanya-tambah. Mode tulis khusus penambahan, termasuk tabel Apache Iceberg: Datastream menambahkan peristiwa perubahan ke tabel tujuan dan mengisi kolom datastream_metadata STRUCT. Untuk menentukan urutan terjadinya peristiwa di sumber, urutkan peristiwa menggunakan kolom datastream_metadata.change_sequence_number atau kolom array datastream_metadata.sort_keys.
Cloud Storage	Di Cloud Storage, peristiwa dari jangka waktu yang sama dapat mencakup lebih dari satu file. Untuk menentukan urutan terjadinya peristiwa di sumber, urutkan peristiwa menggunakan kolom array sort_keys dalam metadata generik.

Konsistensi

Datastream memastikan bahwa data dari database sumber dikirim ke tujuan setidaknya sekali. Tidak ada peristiwa yang terlewat, tetapi ada kemungkinan peristiwa duplikat dalam aliran data. Periode untuk peristiwa duplikat harus dalam hitungan menit, dan ID unik universal (UUID) peristiwa dalam metadata peristiwa dapat digunakan untuk mendeteksi duplikat.

Namun, jangan hanya mengandalkan UUID untuk menghapus duplikat peristiwa di downstream. Jika terjadi gangguan pipeline, Datastream akan membaca ulang log sumber dari checkpoint terakhir yang disimpan untuk memastikan pengiriman. Karena Datastream menghasilkan UUID secara dinamis saat membaca peristiwa, percobaan ulang jenis ini akan menghasilkan UUID baru untuk data sumber yang sama persis. Selain itu, selama operasi massal di database sumber, beberapa baris yang berbeda mungkin memiliki nomor urut perubahan (CSN) yang sama.

Untuk menghapus duplikat data dengan andal, partisi peristiwa berdasarkan kunci utama tabel sumber, dan kelompokkan menggunakan nomor urut perubahan (CSN) (kolom change_sequence_number). Hal ini memastikan bahwa percobaan ulang peristiwa yang sama diciutkan dengan benar sekaligus mempertahankan pembaruan baris yang benar-benar berbeda dari transaksi massal.

Jika file log database berisi transaksi yang belum di-commit, dan jika ada transaksi yang di-roll back, database akan mencerminkan hal ini dalam file log sebagai operasi bahasa pengolahan data (DML) "terbalik". Misalnya, operasi INSERT yang di-roll back akan memiliki operasi DELETE yang sesuai. Datastream membaca operasi ini dari file log.

Tentang aliran data

Setiap aliran data memiliki metadata yang menjelaskan aliran data dan sumber tempat aliran data mengambil data. Metadata ini mencakup informasi seperti nama aliran data, serta profil koneksi sumber dan tujuan.

Untuk melihat definisi lengkap objek Stream, lihat dokumentasi Referensi API.

Status dan kondisi streaming

Streaming dapat berada dalam salah satu status berikut:

• Not started
• Starting
• Running
• Draining
• Paused
• Failed
• Failed permanently

Untuk mengetahui informasi selengkapnya tentang status ini, termasuk cara penggunaannya dalam siklus proses streaming, lihat Siklus proses streaming.

Anda dapat menggunakan log untuk menemukan informasi status tambahan, seperti pengisian ulang tabel atau jumlah baris yang diproses. Anda juga dapat menggunakan FetchStreamErrors API untuk mengambil error.

Metadata objek yang tersedia menggunakan API penemuan

API penemuan menampilkan objek yang mewakili struktur objek yang ditentukan dalam sumber data atau tujuan yang diwakili oleh profil koneksi. Setiap objek memiliki metadata pada objek itu sendiri, serta untuk setiap kolom data yang ditarik. Metadata ini tersedia menggunakan discover API.

Langkah berikutnya

• Untuk mempelajari lebih lanjut streaming, lihat Siklus proses streaming.
• Untuk mempelajari cara membuat aliran data, lihat Membuat aliran data.