stream data real-time ke PolarDB PostgreSQL (Oracle) melalui Flink Sink - Realtime Compute for Apache Flink

Topik ini menjelaskan cara menggunakan konektor PolarDB for PostgreSQL (Kompatibel dengan Oracle) 1.0.

Penting

Konektor PolarDB for PostgreSQL (Kompatibel dengan Oracle) 1.0 akan ditinggalkan pada rilis mendatang. Opsi untuk konektor ini akan dihapus dari Konsol dan tidak akan lagi dipelihara atau diperbarui. Untuk informasi selengkapnya, lihat Akhir Dukungan (EOS) untuk konektor PolarDB for PostgreSQL (Kompatibel dengan Oracle) 1.0. Kami menyarankan Anda segera memigrasikan beban kerja Anda untuk mencegah gangguan layanan.

Informasi latar belakang

PolarDB for PostgreSQL (Kompatibel dengan Oracle) adalah Cloud-Native Database generasi berikutnya yang dikembangkan oleh Alibaba Cloud. Dibangun di atas arsitektur pemisahan storage dan compute, layanan ini menyediakan database yang elastis, berkinerja tinggi, sangat scalable, aman, andal, serta sangat kompatibel dengan Oracle.

Tabel berikut menjelaskan kemampuan konektor PolarDB for PostgreSQL (Kompatibel dengan Oracle) 1.0.

Kategori	Deskripsi
Tipe yang didukung	Sink table
Mode eksekusi	Streaming dan batch
Format data	N/A
Metrik konektor	Untuk sink table: numRecordsOut numRecordsOutPerSecond numBytesOut numBytesOutPerSecond currentSendTime Catatan Untuk informasi selengkapnya mengenai metrik ini, lihat Metrik pemantauan.
Tipe API	SQL
Dukungan update dan delete	Didukung

Prasyarat

Buat kluster dan tabel PolarDB for PostgreSQL (Kompatibel dengan Oracle) 1.0. Untuk informasi selengkapnya, lihat Buat kluster PolarDB for PostgreSQL (Kompatibel dengan Oracle) dan Buat tabel.
Konfigurasikan daftar putih untuk kluster tersebut. Untuk informasi selengkapnya, lihat Konfigurasikan daftar putih kluster.

Batasan

Konektor PolarDB for PostgreSQL (Kompatibel dengan Oracle) 1.0 hanya mendukung PolarDB 1.0 dan tidak mendukung PolarDB 2.0. Untuk PolarDB 2.0, kami menyarankan menggunakan konektor JDBC.
Konektor ini hanya didukung di Ververica Runtime (VVR) 8.0.5 dan versi yang lebih baru.

Sintaks

CREATE TABLE polardbo_table (
 id INT,
 len INT,
 content VARCHAR,
 PRIMARY KEY(id)
) WITH (
 'connector'='polardbo',
 'url'='jdbc:postgresql://<Address>:<PortId>/<DatabaseName>',
 'tableName'='<yourDatabaseTableName>',
 'userName'='<yourDatabaseUserName>',
 'password'='<yourDatabasePassword>'
);

Parameter

Parameter	Deskripsi	Tipe	Wajib	Bawaan	Keterangan
connector	Jenis konektor.	String	Ya	N/A	Harus diatur ke `polardbo`.
url	URL JDBC database.	String	Ya	N/A	Format: `jdbc:postgresql://<Address>:<PortId>/<DatabaseName>`.
tableName	Nama tabel database.	String	Ya	N/A	N/A
userName	Username untuk koneksi ke database.	String	Ya	N/A	N/A
password	Password untuk koneksi ke database.	String	Ya	N/A	Untuk meningkatkan keamanan, gunakan Project Variables alih-alih hardcoding kredensial dalam teks biasa. Untuk informasi selengkapnya, lihat Manajemen variabel.
maxRetryTimes	Jumlah maksimum percobaan ulang untuk operasi penulisan yang gagal.	Integer	Tidak	3	N/A
targetSchema	Nama skema.	String	Tidak	public	N/A
caseSensitive	Menentukan apakah nama tabel dan bidang bersifat case-sensitive.	String	Tidak	false	Nilai yang valid: `true`: Case-sensitive. `false`: Tidak case-sensitive.
connectionMaxActive	Jumlah maksimum koneksi aktif dalam kolam koneksi.	Integer	Tidak	5	Sistem secara otomatis melepaskan koneksi idle ke layanan database. Penting Mengatur parameter ini ke nilai yang terlalu besar dapat membebani server dengan koneksi.
retryWaitTime	Interval antar percobaan ulang.	Integer	Tidak	100	Unit: milidetik.
batchSize	Jumlah catatan yang ditulis dalam satu batch.	Integer	Tidak	500	N/A
flushIntervalMs	Interval flush. Jika jumlah catatan yang dicache tidak mencapai `batchSize` dalam interval ini, sistem akan melakukan flush semua data dari cache.	Integer	Tidak	N/A	Unit: milidetik.
writeMode	Mode penulisan.	String	Tidak	insert	Nilai yang valid: `insert`: Langsung memasukkan data. Jika terjadi konflik, perilaku ditentukan oleh parameter `conflictMode`. `upsert`: Melakukan update jika terjadi konflik. Mode ini hanya berlaku untuk tabel yang memiliki Primary Key.
conflictMode	Kebijakan penanganan konflik Primary Key atau unique index selama operasi `insert`.	String	Tidak	strict	Nilai yang valid: `strict`: Melemparkan error saat terjadi konflik. `ignore`: Mengabaikan catatan yang konflik. `update`: Memperbarui catatan saat terjadi konflik. Ini dapat digunakan untuk tabel tanpa Primary Key, tetapi performanya lebih rendah.

Pemetaan tipe data

Tabel berikut memetakan tipe data Flink ke tipe data PolarDB for PostgreSQL (Kompatibel dengan Oracle) 1.0 saat konektor digunakan untuk Sink table.

Tipe PolarDB	Tipe Flink
BOOLEAN	BOOLEAN
INT	INT
NUMBER	BIGINT
NUMBER	DOUBLE
VARCHAR	VARCHAR
TIMESTAMP	TIMESTAMP
VARCHAR	DATE

Contoh

Sink table

CREATE TEMPORARY TABLE datagen_source (
 `name` VARCHAR,
 `age` INT
)
COMMENT 'tabel sumber datagen'
WITH (
 'connector' = 'datagen'
);

CREATE TABLE polardbo_sink (
 name VARCHAR,
 age INT
) WITH (
 'connector'='polardbo',
 'url'='jdbc:postgresql://<Address>:<PortId>/<DatabaseName>',
 'tableName'='<yourDatabaseTableName>',
 'userName'='<yourDatabaseUserName>',
 'password'='<yourDatabasePassword>'
);

INSERT INTO polardbo_sink
SELECT * FROM datagen_source;