Menulis dan mengonsumsi data tabel Paimon - Realtime Compute for Apache Flink

Topik ini menjelaskan cara menyisipkan, memperbarui, menimpa, atau menghapus data dalam tabel Paimon menggunakan Konsol Realtime Compute for Apache Flink, serta cara mengonsumsi data dari tabel tersebut dan menentukan offset konsumsi.

Prasyarat

Anda harus terlebih dahulu membuat katalog Paimon dan tabel Paimon. Untuk informasi selengkapnya, lihat Mengelola katalog Paimon.

Batasan

Tabel Paimon hanya didukung di Ververica Runtime (VVR) 8.0.5 dan versi yang lebih baru.

Menulis data ke tabel Paimon

Menyinkronkan data dan skema dengan CTAS/CDAS

Untuk informasi selengkapnya, lihat Mengelola katalog Paimon.

Menyisipkan atau memperbarui data dengan INSERT INTO

Anda dapat menggunakan pernyataan INSERT INTO untuk langsung menyisipkan atau memperbarui data dalam tabel Paimon.

Tabel primary key Paimon dapat menerima pesan semua jenis, termasuk INSERT, UPDATE_BEFORE, UPDATE_AFTER, dan DELETE. Tabel ini menggabungkan data dengan primary key yang sama saat penulisan berdasarkan mekanisme penggabungan data.
Tabel append-only Paimon, juga dikenal sebagai tabel non-primary key, hanya menerima pesan bertipe INSERT.

Menimpa data dengan INSERT OVERWRITE

Operasi penimpaan akan menghapus data yang ada dan menulis data baru. Anda dapat menggunakan pernyataan INSERT OVERWRITE untuk menimpa seluruh tabel Paimon atau partisi tertentu. Kode berikut memberikan contohnya.

Catatan

Pernyataan INSERT OVERWRITE hanya didukung dalam pekerjaan batch.
Secara default, operasi INSERT OVERWRITE tidak menghasilkan data changelog. Pekerjaan streaming downstream tidak dapat mengonsumsi data yang dihapus dan diimpor. Jika Anda perlu mengonsumsi jenis data ini, lihat Streaming dan konsumsi hasil pernyataan INSERT OVERWRITE.

Timpa seluruh tabel non-partisi my_table.
```
INSERT OVERWRITE my_table SELECT ...;
```

Timpa partisi dt=20240108,hh=06 dalam tabel my_table.

INSERT OVERWRITE my_table PARTITION (`dt` = '20240108', `hh` = '06') SELECT ...;

Timpa partisi secara dinamis dalam tabel my_table. Partisi yang muncul dalam hasil pernyataan SELECT akan ditimpa, sedangkan partisi lain tetap tidak berubah.
```
INSERT OVERWRITE my_table SELECT ...;
```

Timpa seluruh tabel partisi my_table.

INSERT OVERWRITE my_table /*+ OPTIONS('dynamic-partition-overwrite' = 'false') */ SELECT ...;

Menghapus data menggunakan DELETE

Anda dapat menggunakan pernyataan DELETE untuk menghapus data dari tabel primary key Paimon. Anda hanya dapat mengeksekusi pernyataan DELETE di Data Exploration.

-- Hapus semua data dengan currency = 'UNKNOWN' dari tabel my_table.
DELETE FROM my_table WHERE currency = 'UNKNOWN';

Memfilter pesan delete

Saat menggunakan tabel primary key Paimon, pesan DELETE secara default menghapus data dengan primary key yang sesuai. Untuk mencegah tabel Paimon memproses pesan-pesan ini, gunakan Petunjuk SQL untuk mengatur parameter berikut ke true. Hal ini akan memfilter pesan delete.

Parameter	Deskripsi	Tipe	Default
ignore-delete	Menentukan apakah akan memfilter pesan delete.	Boolean	false

Menyesuaikan parallelisme sink

Anda dapat menggunakan Petunjuk SQL untuk mengatur parameter berikut guna menyesuaikan parallelisme operator sink.

Parameter	Deskripsi	Type	Default
sink.parallelism	Mengatur parallelisme operator sink Paimon.	Integer	None

Sebagai contoh, pernyataan SQL berikut mengatur parallelisme operator sink Paimon menjadi 10.

INSERT INTO t /*+ OPTIONS('sink.parallelism' = '10') */ SELECT * FROM s;

Mengonsumsi data dari tabel Paimon

Pekerjaan streaming

Catatan

Untuk tabel primary key Paimon yang dikonsumsi oleh pekerjaan streaming, Anda harus mengonfigurasi changelog producer.

Secara default, operator source Paimon dalam pekerjaan streaming pertama kali menghasilkan seluruh data tabel saat pekerjaan dimulai, lalu data inkremental mulai dari titik tersebut.

Mengonsumsi data dari offset tertentu

Anda dapat mengonsumsi data dari offset tertentu dengan salah satu cara berikut:

Jika Anda tidak perlu mengonsumsi seluruh data dalam tabel Paimon saat startup pekerjaan dan hanya ingin mengonsumsi data inkremental berikutnya, Anda dapat mengatur 'scan.mode' = 'latest' melalui Petunjuk SQL.
```
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'latest') */;
```
Jika Anda tidak ingin mengonsumsi seluruh data dan hanya ingin mengonsumsi data inkremental dari titik waktu tertentu, Anda dapat menggunakan Petunjuk SQL untuk mengatur parameter scan.timestamp-millis. Nilai parameter ini merepresentasikan jumlah milidetik sejak Unix Epoch (1970-01-01 00:00:00 UTC) hingga titik waktu yang ditentukan.
```
SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;
```
Untuk mengonsumsi data yang ditulis setelah waktu tertentu dan kemudian terus mengonsumsi data inkremental berikutnya, gunakan salah satu metode berikut.

Catatan
Metode konsumsi ini membaca file data yang dimodifikasi setelah waktu yang ditentukan. Karena proses compaction, file data mungkin berisi sedikit data yang ditulis sebelum waktu yang ditentukan. Anda dapat menambahkan klausa WHERE pada pekerjaan SQL Anda untuk memfilter data sesuai kebutuhan.
- Jangan atur Petunjuk SQL apa pun. Saat memulai pekerjaan, Anda dapat memilih Specify source's start time. Di kotak dialog Job Start, pilih Stateless start-up, aktifkan sakelar Specify source's start time, lalu atur waktu target.
- Atur parameter scan.file-creation-time-millis menggunakan Petunjuk SQL.
```
SELECT * FROM t /*+ OPTIONS('scan.file-creation-time-millis' = '1678883047356') */;
```
Jika Anda tidak ingin mengonsumsi seluruh data dan hanya ingin mengonsumsi data inkremental mulai dari file snapshot tertentu, Anda dapat menggunakan Petunjuk SQL untuk mengatur parameter scan.snapshot-id. Nilai parameter ini adalah ID file snapshot yang ditentukan.
```
SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;
```
Jika Anda ingin mengonsumsi seluruh data dari file snapshot tertentu dan melanjutkan konsumsi data inkremental, Anda dapat menggunakan Petunjuk SQL untuk mengatur parameter 'scan.mode' = 'from-snapshot-full' dan scan.snapshot-id. Nilai parameter scan.snapshot-id adalah ID file snapshot yang ditentukan.
```
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'from-snapshot-full', 'scan.snapshot-id' = '1') */;
```

Tentukan ID konsumen

Consumer ID menyimpan progres konsumsi untuk tabel Paimon. Ini terutama digunakan dalam skenario berikut:

Menetapkan consumer ID menyimpan progres konsumsinya ke file metadata tabel Paimon. Hal ini memungkinkan pekerjaan melanjutkan konsumsi dari titik interupsi, bahkan jika nantinya dijalankan dalam mode tanpa status.
Setelah Anda menetapkan consumer ID, snapshot yang belum dikonsumsi tidak akan dihapus karena kedaluwarsa. Hal ini mencegah error yang disebabkan oleh kecepatan konsumsi yang tidak mampu mengimbangi kecepatan kedaluwarsa snapshot.

Anda dapat mengatur parameter consumer-id untuk memberikan Consumer ID kepada operator source Paimon dalam pekerjaan streaming. Nilai Consumer ID dapat berupa string apa pun. Saat Consumer ID dibuat pertama kali, offset awalnya ditentukan sesuai aturan dalam Mengonsumsi Tabel Paimon dari Offset Tertentu. Anda kemudian dapat melanjutkan konsumsi tabel Paimon dengan menggunakan kembali Consumer ID yang sama.

Sebagai contoh, pernyataan SQL berikut menetapkan consumer ID bernama test-id untuk operator source Paimon. Jika Anda ingin mengatur ulang offset untuk consumer ID tertentu, Anda juga dapat mengatur 'consumer.ignore-progress' = 'true'.

SELECT * FROM t /*+ OPTIONS('consumer-id' = 'test-id') */;

Catatan

File snapshot yang belum dikonsumsi oleh consumer ID tidak akan dihapus saat kedaluwarsa. Jika consumer ID usang tidak dibersihkan, file snapshot dan file data historis yang bersangkutan tidak akan pernah dihapus dan akan mengonsumsi storage space. Anda dapat mengatur parameter tabel consumer.expiration-time untuk membersihkan consumer ID yang tidak digunakan selama periode tertentu. Sebagai contoh, 'consumer.expiration-time' = '3d' menunjukkan bahwa consumer ID yang tidak digunakan selama 3 hari akan dibersihkan.

Streaming dan mengonsumsi hasil INSERT OVERWRITE

Secara default, operasi INSERT OVERWRITE tidak menghasilkan data changelog, sehingga data yang dihapus dan diimpor tidak dapat dikonsumsi oleh pekerjaan streaming downstream. Jika Anda perlu mengonsumsi data tersebut, Anda dapat mengatur 'streaming-read-overwrite' = 'true' dalam pekerjaan konsumsi streaming menggunakan Petunjuk SQL.

SELECT * FROM t /*+ OPTIONS('streaming-read-overwrite' = 'true') */;

Pekerjaan batch

Secara default, operator source Paimon dalam pekerjaan batch membaca snapshot terbaru dan mengeluarkan data status terbaru dari tabel Paimon.

Perjalanan waktu batch

Anda dapat mengkueri status tabel Paimon pada titik waktu tertentu dengan mengatur parameter scan.timestamp-millis dalam Petunjuk SQL. Nilai parameter ini merepresentasikan jumlah milidetik sejak Unix Epoch (1970-01-01 00:00:00 UTC) hingga waktu yang ditentukan.

SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;

Anda dapat mengkueri status tabel Paimon pada saat snapshot dibuat dengan menggunakan Petunjuk SQL untuk mengatur parameter scan.snapshot-id ke ID snapshot yang ditentukan.

SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;

Mengkueri perubahan antar snapshot

Jika Anda ingin mengkueri perubahan data dalam tabel Paimon antara dua snapshot, Anda dapat mengatur parameter incremental-between menggunakan Petunjuk SQL. Sebagai contoh, untuk melihat semua data yang berubah antara snapshot 20 dan snapshot 12, pernyataan SQL-nya adalah sebagai berikut.

SELECT * FROM t /*+ OPTIONS('incremental-between' = '12,20') */;

Catatan

Karena pekerjaan batch tidak mendukung konsumsi pesan DELETE, pesan-pesan ini secara default dibuang. Jika Anda ingin mengonsumsi pesan DELETE dalam pekerjaan batch, kueri tabel sistem Audit Log. Contoh: SELECT * FROM `t$audit_log ` /*+ OPTIONS('incremental-between' = '12,20') */;.

Sesuaikan paralelisme sumber

Secara default, Paimon secara otomatis melakukan inferensi parallelisme operator source berdasarkan informasi seperti jumlah partisi dan bucket. Anda dapat menggunakan Petunjuk SQL untuk mengatur parameter berikut guna menyesuaikan parallelisme.

Parameter	Tipe	Default	Deskripsi
scan.parallelism	Integer	None	Mengatur parallelisme operator source Paimon.
scan.infer-parallelism	Boolean	true	Menentukan apakah akan secara otomatis melakukan inferensi parallelisme operator source Paimon.
scan.infer-parallelism.max	Integer	1024	Batas atas parallelisme yang diinferensi secara otomatis untuk operator source Paimon.

Pernyataan SQL berikut memberikan contoh cara mengatur parallelisme operator source Paimon menjadi 10.

SELECT * FROM t /*+ OPTIONS('scan.parallelism' = '10') */;

Menggunakan tabel Paimon sebagai tabel dimensi

Tabel Paimon juga dapat digunakan sebagai tabel dimensi. Untuk informasi selengkapnya tentang sintaks JOIN tabel dimensi, lihat Pernyataan JOIN tabel dimensi.

Menulis dan mengonsumsi tipe VARIANT

Di Ververica Runtime (VVR) 11.1 dan versi yang lebih baru, tabel Paimon mendukung tipe data semi-terstruktur VARIANT. Tipe ini memungkinkan Anda mengonversi string JSON VARCHAR ke tipe VARIANT menggunakan PARSE_JSON atau TRY_PARSE_JSON. Menulis dan mengonsumsi tipe VARIANT secara langsung secara signifikan meningkatkan kinerja kueri dan pemrosesan JSON.

Kode berikut memberikan contohnya:

CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  k BIGINT,
  info VARIANT
);
INSERT INTO `my-catalog`.`my_db`.`my_tbl` 
SELECT k, PARSE_JSON(jsonStr) FROM T;

Dokumentasi terkait

Saat menulis dan mengonsumsi data dari tabel Paimon, Anda dapat menggunakan Petunjuk SQL untuk sementara memodifikasi parameter tabel. Untuk informasi selengkapnya, lihat Mengelola tabel Paimon.
Untuk informasi selengkapnya tentang fitur dan fungsi dasar tabel primary key Paimon dan tabel append-only Paimon, lihat Tabel primary key dan append-only Paimon.
Untuk informasi selengkapnya tentang optimasi umum untuk tabel primary key Paimon dan tabel Append Scalable dalam berbagai skenario, lihat Optimasi kinerja Paimon.
Konsumsi data dari tabel Paimon bergantung pada file snapshot. Jika waktu kedaluwarsa snapshot terlalu singkat atau pekerjaan konsumsi tidak efisien, file snapshot yang sedang dikonsumsi mungkin dihapus setelah kedaluwarsa, yang menyebabkan pekerjaan konsumsi melaporkan error File xxx not found, Possible causes. Untuk solusinya, lihat Mengatasi error "File xxx not found, Possible causes" dalam pekerjaan baca Paimon.