StarRocks SQL connector - Realtime Compute for Apache Flink

Pelajari cara menggunakan connector StarRocks dalam pekerjaan SQL.

Latar Belakang

StarRocks adalah gudang data Pemrosesan Paralel Masif (MPP) generasi berikutnya yang menawarkan performa sangat cepat di semua skenario serta pengalaman analitik terpadu. StarRocks memiliki keunggulan berikut:

StarRocks kompatibel dengan protokol MySQL, sehingga Anda dapat menggunakan klien MySQL dan alat Business Intelligence (BI) umum untuk menghubungkan dan menganalisis data.
StarRocks menggunakan arsitektur terdistribusi:
- Tabel data dipartisi secara horizontal dan disimpan dalam beberapa replika.
- Kluster dapat diskalakan secara fleksibel dan mampu menganalisis hingga 10 petabyte (PB) data.
- Menggunakan framework MPP untuk mempercepat komputasi paralel.
- Mendukung beberapa replika untuk menyediakan toleransi kesalahan.

Connector Flink menyimpan data dalam cache dan menggunakan Stream Load untuk menuliskannya secara batch ke tabel sink. Connector ini membaca dari tabel sumber dengan mengambil data secara batch. Tabel berikut mencantumkan kemampuan connector StarRocks.

Kategori	Deskripsi
Jenis yang didukung	tabel sumber, tabel dimensi, tabel sink, dan target ingesti data
Mode eksekusi	mode streaming dan mode batch
Format data	CSV
Metrik khusus connector	Tidak ada
Jenis API	DataStream, SQL, dan YAML untuk ingesti data
Dukungan pembaruan/penghapusan pada tabel sink	Ya

Prasyarat

Anda harus memiliki kluster StarRocks yang dideploy di EMR atau yang dikelola sendiri di ECS.

Batasan

Hanya Ververica Runtime (VVR) 11.1 atau versi lebih baru yang mendukung penggabungan dengan tabel dimensi.
Untuk menghindari pembatasan akses jaringan, tambahkan port kluster StarRocks berikut ke daftar putih security group atau firewall: 9030, 8030, 8040, 9060, 8060, 9020.
Saat tabel StarRocks target berisi kolom tersembunyi yang di-generate, deklarasikan hanya kolom fisik aktual yang akan ditulis dalam pekerjaan Flink, dan kecualikan bidang kolom yang di-generate. Kolom tersembunyi yang di-generate secara otomatis oleh ekspresi partisi StarRocks (seperti __generated_partition_column_0) tidak menerima penulisan eksternal. Secara default, connector membuat permintaan tulis berdasarkan skema lengkap, yang menyebabkan pekerjaan gagal. Untuk informasi lebih lanjut tentang kolom yang di-generate di StarRocks, lihat Generated columns.

Fitur

StarRocks pada E-MapReduce mendukung pernyataan CREATE TABLE AS SELECT (CTAS) dan CREATE DATABASE AS SELECT (CDAS). CTAS menyinkronkan skema dan data satu tabel, sedangkan CDAS menyinkronkan seluruh database atau beberapa tabel dalam database yang sama. Untuk informasi lebih lanjut, lihat Gunakan pernyataan CTAS dan CDAS di Realtime Compute for Apache Flink untuk menyinkronkan data dari database MySQL ke StarRocks.

Sintaks

CREATE TABLE USER_RESULT(
 name VARCHAR,
 score BIGINT
 ) WITH (
 'connector' = 'starrocks',
 'jdbc-url'='jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx',
 'load-url'='fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port',
 'database-name' = 'xxx',
 'table-name' = 'xxx',
 'username' = 'xxx',
 'password' = 'xxx'
 );

Parameter

Jenis	Parameter	Deskripsi	Tipe	Wajib	Default	Keterangan
Umum	connector	Menentukan connector yang digunakan.	String	Ya	—	Nilainya harus `starrocks`.
	jdbc-url	URL Java Database Connectivity (JDBC).	String	Ya	—	Tentukan alamat IP dan port JDBC FE dalam format `jdbc:mysql://ip:port`.
	database-name	Nama database StarRocks.	String	Ya	—	—
	table-name	Nama tabel StarRocks.	String	Ya	—	—
	username	Username untuk menghubungkan ke StarRocks.	String	Ya	—	—
	password	Password untuk menghubungkan ke StarRocks.	String	Ya	—	—
	starrocks.create.table.properties	Menentukan properti untuk pembuatan tabel otomatis.	String	Tidak	—	Menentukan properti awal tabel, seperti engine dan jumlah replika. Contoh: 'starrocks.create.table.properties' = 'buckets 8' atau 'starrocks.create.table.properties' = 'replication_num=1'.
Khusus Sumber	scan-url	URL pemindaian data.	String	Tidak	—	Menentukan alamat IP dan port HTTP FE. Format: `fe_ip:http_port;fe_ip:http_port`. Catatan Untuk menentukan beberapa alamat IP dan port, pisahkan dengan koma (,).
	scan.connect.timeout-ms	Timeout untuk `flink-connector-starrocks` menghubungkan ke StarRocks. Connector melaporkan error jika koneksi tidak berhasil dibuat dalam batas waktu ini.	String	Tidak	1000	Unit: milidetik.
	scan.params.keep-alive-min	Durasi keep-alive untuk tugas kueri.	String	Tidak	10	—
	scan.params.query-timeout-s	Timeout untuk tugas kueri. Jika tidak ada hasil yang dikembalikan dalam periode ini, sistem menghentikan tugas kueri.	String	Tidak	600	Unit: detik.
	scan.params.mem-limit-byte	Batas memori untuk satu kueri pada node BE.	String	Tidak	1073741824 (1 GB)	Unit: byte.
	scan.max-retries	Jumlah maksimum percobaan ulang untuk kueri yang gagal. Connector melaporkan error jika batas ini terlampaui.	String	Tidak	1	—
Khusus Sink	load-url	URL impor data.	String	Ya	—	Tentukan alamat IP dan port HTTP FE dalam format `fe_ip:http_port;fe_ip:http_port`. Catatan Untuk menentukan beberapa alamat IP dan port, pisahkan dengan titik koma (;).
	sink.semantic	Semantik pengiriman untuk penulisan.	String	Tidak	at-least-once	Nilai yang valid: `at-least-once` (default): Menjamin data dikirimkan setidaknya sekali. `exactly-once`: Menjamin data dikirimkan tepat satu kali.
	sink.buffer-flush.max-bytes	Jumlah maksimum data yang dibuffer sebelum flush.	String	Tidak	94371840 (90 MB)	Rentang valid: 64 MB hingga 10 GB.
	sink.buffer-flush.max-rows	Jumlah maksimum baris yang dibuffer sebelum flush.	String	Tidak	500000	Rentang valid: 1.000 hingga 5.000.000.
	sink.buffer-flush.interval-ms	Interval flush buffer.	String	Tidak	300000	Rentang valid: 1.000 ms hingga 3.600.000 ms.
	sink.max-retries	Jumlah maksimum percobaan ulang untuk penulisan yang gagal.	String	Tidak	3	Rentang valid: 0 hingga 1000.
	sink.connect.timeout-ms	Timeout untuk menghubungkan ke StarRocks.	String	Tidak	1000	Rentang valid: 100 hingga 60.000. Unit: milidetik.
	sink.properties.*	Properti Stream Load tambahan untuk sink.	String	Tidak	—	Parameter ini mengontrol perilaku Stream Load. Misalnya, `sink.properties.format` menentukan format data yang diimpor, seperti CSV. Untuk parameter lainnya, lihat Stream Load.
Khusus Dimensi	lookup.cache.enabled	Menentukan apakah caching untuk tabel dimensi diaktifkan.	Boolean	Tidak	true	Nilai yang valid: `true`: Mengaktifkan caching. Setelah data tabel dibaca pertama kali, sistem menyimpannya dalam memori. Permintaan berikutnya menggunakan data cache dalam periode validitasnya untuk mengurangi overhead I/O. `false`: Menonaktifkan caching. Setiap kueri langsung mengakses sumber data. Penting Fitur ini memerlukan engine Realtime Compute for Apache Flink VVR 11.1 dan versi lebih baru. Kami merekomendasikan menonaktifkan fitur ini dalam skenario berikut: Data tabel dimensi sering diperbarui, dan data real-time diperlukan. Tabel berisi banyak data, sehingga berisiko menyebabkan overflow memori.

Pemetaan tipe data

Tipe data StarRocks	Tipe data Flink
NULL	NULL
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
BIGINT UNSIGNED Catatan Memerlukan engine Realtime Compute for Apache Flink VVR 8.0.10 dan versi lebih baru.	DECIMAL(20,0)
LARGEINT	DECIMAL(20,0)
FLOAT	FLOAT
DOUBLE	DOUBLE
DATE	DATE
DATETIME	TIMESTAMP
DECIMAL	DECIMAL
DECIMALV2	DECIMAL
DECIMAL32	DECIMAL
DECIMAL64	DECIMAL
DECIMAL128	DECIMAL
CHAR(m) Catatan VVR 8.0.10 secara otomatis memperpanjang panjang CHAR tiga kali lipat (m=n3, dengan n<=85) untuk mengakomodasi perbedaan encoding antara MySQL dan StarRocks. VVR 8.0.11 dan versi lebih baru secara otomatis memperpanjang panjang CHAR empat kali lipat (m=n4, dengan n<=63) untuk mengakomodasi perbedaan encoding antara MySQL dan StarRocks. Panjang maksimum untuk tipe CHAR StarRocks adalah 255. Oleh karena itu, Flink memetakan tipe CHAR ke tipe CHAR StarRocks hanya jika panjang yang diperpanjang secara otomatis tidak melebihi 255.	CHAR(n)
VARCHAR(m) Catatan VVR 8.0.10 secara otomatis memperpanjang panjang VARCHAR tiga kali lipat (m=n3, dengan n>85) untuk mengakomodasi perbedaan encoding antara MySQL dan StarRocks. VVR 8.0.11 dan versi lebih baru secara otomatis memperpanjang panjang VARCHAR empat kali lipat (m=n4, dengan n>63) untuk mengakomodasi perbedaan encoding antara MySQL dan StarRocks. Panjang maksimum untuk tipe CHAR StarRocks adalah 255. Oleh karena itu, jika panjang yang diperpanjang secara otomatis dari tipe CHAR Flink melebihi 255, Flink memetakan tipe tersebut ke tipe VARCHAR StarRocks.	CHAR(n)
VARCHAR	STRING
VARBINARY Catatan Memerlukan engine Realtime Compute for Apache Flink VVR 8.0.10 dan versi lebih baru.	VARBINARY

Contoh

Tabel sumber, tabel sink, dan tabel dimensi dalam contoh berikut dapat digunakan bersama. Tabel runoob_tbl_source yang dideklarasikan dalam contoh tabel sumber juga berfungsi sebagai sumber data untuk contoh tabel sink dan tabel dimensi. Sebelum menggunakan contoh-contoh ini, siapkan tabel dengan tipe bidang yang sesuai di StarRocks, lalu ganti titik akhir, nama database dan tabel, serta variabel rahasia.

Tabel sumber

CREATE TEMPORARY TABLE runoob_tbl_source (
    runoob_id BIGINT NOT NULL,
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    proc_time AS PROCTIME() --Kolom waktu pemrosesan, digunakan oleh temporal join dalam contoh tabel dimensi
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --Titik akhir protokol MySQL FE. Pisahkan beberapa titik akhir dengan koma (,).
    'scan-url' = '<fe_host>:8030', --Titik akhir HTTP FE. Pisahkan beberapa titik akhir dengan koma (,).
    'database-name' = '<database_name>',
    'table-name' = '<source_table_name>',
    'username' = '${secret_values.starrocks_username}', --Gunakan variabel rahasia untuk mencegah kebocoran kredensial
    'password' = '${secret_values.starrocks_password}',
    'scan.params.query-timeout-s' = '600' --Timeout untuk satu kueri, dalam detik
);

Catatan

Tabel sumber dibaca dalam mode batch. Setelah pekerjaan dimulai, tabel melakukan pemindaian data saat ini lalu berhenti menghasilkan output. Tabel ini tidak terus-menerus menghasilkan perubahan inkremental.

Tabel sink

Catatan

StarRocks mengizinkan kolom kunci primer menjadi NULLABLE. Namun, Flink tidak mendukung kunci primer yang berisi kolom nullable. Model konsistensi data Flink mensyaratkan bahwa kunci primer harus unik dan non-nullable. Jika tidak, Flink akan melemparkan error Invalid primary key. Column 'xxx' is nullable. Untuk informasi lebih lanjut, lihat Error "Invalid primary key. Column 'xxx' is nullable.".

VVR 11+

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, --Kolom kunci primer harus dideklarasikan sebagai NOT NULL
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --Titik akhir protokol MySQL FE. Pisahkan beberapa titik akhir dengan koma (,).
    'load-url' = '<fe_host>:8030', --Titik akhir HTTP FE. Pisahkan beberapa titik akhir dengan titik koma (;).
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', --Gunakan variabel rahasia untuk mencegah kebocoran kredensial
    'password' = '${secret_values.starrocks_password}',
    'sink.version' = 'V2', --Transactional Stream Load. Memerlukan StarRocks 2.4 atau versi lebih baru.
    'sink.semantic' = 'at-least-once', --Opsional. exactly-once juga didukung dan memerlukan checkpointing.
    'sink.buffer-flush.interval-ms' = '5000' --Interval flush batch, dalam milidetik
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

VVR 8+

Pada VVR 8.x, nilai default sink.version adalah V1, dan V1, V2, serta AUTO didukung. Parameter yang hanya tersedia di VVR 11.x, seperti sink.ignore.update-before, tidak didukung:

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, --Kolom kunci primer harus dideklarasikan sebagai NOT NULL
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --Titik akhir protokol MySQL FE. Pisahkan beberapa titik akhir dengan koma (,).
    'load-url' = '<fe_host>:8030', --Titik akhir HTTP FE. Pisahkan beberapa titik akhir dengan titik koma (;).
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', --Gunakan variabel rahasia untuk mencegah kebocoran kredensial
    'password' = '${secret_values.starrocks_password}',
    'sink.semantic' = 'at-least-once', --Opsional. exactly-once juga didukung dan memerlukan checkpointing.
    'sink.buffer-flush.interval-ms' = '5000', --Interval flush batch, dalam milidetik
    'sink.max-retries' = '3' --Jumlah percobaan ulang setelah kegagalan Stream Load
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

Tabel dimensi

Penggabungan tabel dimensi hanya didukung di VVR 11.1 dan versi lebih baru. Tabel dimensi harus mendeklarasikan kunci primer sebagai kunci gabungan, dan temporal join menggunakan waktu pemrosesan. Contoh berikut membangun tabel runoob_tbl_source dari contoh tabel sumber, yang mencakup kolom waktu pemrosesan proc_time:

CREATE TEMPORARY TABLE sr_dim (
    runoob_id BIGINT NOT NULL,
    runoob_author STRING,
    PRIMARY KEY (runoob_id) NOT ENFORCED --Tabel dimensi harus mendeklarasikan kunci primer sebagai kunci gabungan
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --Titik akhir protokol MySQL FE. Pisahkan beberapa titik akhir dengan koma (,).
    'scan-url' = '<fe_host>:8030', --Titik akhir HTTP FE. Pisahkan beberapa titik akhir dengan koma (,).
    'database-name' = '<database_name>',
    'table-name' = '<dim_table_name>',
    'username' = '${secret_values.starrocks_username}', --Gunakan variabel rahasia untuk mencegah kebocoran kredensial
    'password' = '${secret_values.starrocks_password}',
    'lookup.cache.ttl-ms' = '5000', --TTL cache tabel dimensi, dalam milidetik
    'lookup.cache.enabled' = 'true' --Default: true. Atur ke false untuk mengkueri tabel langsung tanpa cache.
);

SELECT o.runoob_id, o.runoob_title, d.runoob_author
FROM runoob_tbl_source AS o
JOIN sr_dim FOR SYSTEM_TIME AS OF o.proc_time AS d --Temporal join yang menggunakan kolom waktu pemrosesan tabel sumber
ON o.runoob_id = d.runoob_id;