AnalyticDB:Tulis\ data\ ke\ AnalyticDB\ for\ PostgreSQL\ menggunakan\ Realtime\ Compute\ for\ Apache\ Flink

Batasan

• Realtime Compute for Apache Flink tidak dapat membaca data dari AnalyticDB for PostgreSQL dalam mode serverless.

• Hanya Ververica Runtime (VVR) 6.0.0 atau versi yang lebih baru yang mendukung konektor AnalyticDB for PostgreSQL.

• Hanya VVR 8.0.1 atau versi yang lebih baru yang mendukung AnalyticDB for PostgreSQL V7.0.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

• Ruang kerja Flink yang sepenuhnya dikelola. Lihat Aktifkan Flink yang sepenuhnya dikelola.

• Instans AnalyticDB for PostgreSQL. Lihat Buat instans.

• Instans AnalyticDB for PostgreSQL dan ruang kerja Flink yang sepenuhnya dikelola berada dalam virtual private cloud (VPC) yang sama.

Konfigurasi instans AnalyticDB for PostgreSQL

1. Masuk ke Konsol AnalyticDB for PostgreSQL.

2. Tambahkan Blok CIDR dari ruang kerja Flink yang sepenuhnya dikelola ke Daftar putih alamat IP instans tersebut. Lihat Konfigurasi Daftar putih alamat IP.

3. Klik Log On to Database. Lihat Koneksi klien untuk opsi koneksi.

4. Buat tabel tujuan:

   CREATE TABLE test_adbpg_table(
     b1 int,
     b2 int,
     b3 text,
     PRIMARY KEY(b1)
   );

Siapkan konektor Flink

Konektor AnalyticDB for PostgreSQL didistribusikan sebagai file JAR yang dihosting di GitHub. Unduh file tersebut dan unggah ke ruang kerja Flink Anda sebagai konektor kustom.

1. Masuk ke Konsol Realtime Compute for Apache Flink. Pada tab Fully Managed Flink, temukan ruang kerja Anda dan klik Console di kolom Actions.

2. Pada panel navigasi sebelah kiri, klik Connectors.

3. Klik Create Custom Connector dan unggah file JAR.

   Catatan Dapatkan file JAR dari rilis GitHub. Versi JAR harus sesuai dengan versi mesin Flink ruang kerja Anda.

4. Klik Next. Sistem akan mengurai konten JAR. Jika penguraian gagal, periksa apakah kode konektor sesuai dengan standar komunitas Apache Flink.

5. Klik Finish. Konektor akan muncul dalam daftar konektor.

Buat pekerjaan Flink

1. Masuk ke Konsol Realtime Compute for Apache Flink. Pada tab Fully Managed Flink, temukan ruang kerja Anda dan klik Console di kolom Actions.

2. Pada panel navigasi sebelah kiri, klik SQL Editor. Di pojok kiri atas, klik New.

3. Pada kotak dialog New Draft, pada tab SQL Scripts, klik Blank Stream Draft lalu klik Next.

4. Konfigurasi draft dan klik Create.

   Parameter	Deskripsi	Contoh
   Name	Nama unik untuk draft dalam proyek saat ini	adbpg-test
   Location	Folder tempat file kode disimpan	Draft
   Engine Version	Versi mesin Flink. Lihat Versi mesin untuk versi yang didukung	vvr-6.0.7-flink-1.15

Tulis data ke AnalyticDB for PostgreSQL

-- Source: generates random data using the built-in datagen connector
CREATE TABLE datagen_source (
  f_sequence INT,
  f_random INT,
  f_random_str STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '5',
  'fields.f_sequence.kind' = 'sequence',
  'fields.f_sequence.start' = '1',
  'fields.f_sequence.end' = '1000',
  'fields.f_random.min' = '1',
  'fields.f_random.max' = '1000',
  'fields.f_random_str.length' = '10'
);

-- Sink: writes data to AnalyticDB for PostgreSQL via the adbpg connector
CREATE TABLE test_adbpg_table (
  `B1` bigint,
  `B2` bigint,
  `B3` VARCHAR,
  `B4` VARCHAR,
  PRIMARY KEY(B1) NOT ENFORCED
) WITH (
  'connector' = 'adbpg-nightly-1.13',
  'url' = 'jdbc:postgresql://<internal-endpoint>:5432/<database-name>',
  'tablename' = 'test_adbpg_table',
  'username' = '<username>',
  'password' = '<password>',
  'maxretrytimes' = '2',
  'batchsize' = '50000',
  'connectionmaxactive' = '5',
  'conflictmode' = 'ignore',
  'usecopy' = '0',
  'targetschema' = 'public',
  'exceptionmode' = 'ignore',
  'casesensitive' = '0',
  'writemode' = '1',
  'retrywaittime' = '200'
);

Verifikasi hasil

1. Sambungkan ke database AnalyticDB for PostgreSQL. Lihat Koneksi klien.

2. Kueri tabel untuk memastikan data sedang ditulis:

   SELECT * FROM test_adbpg_table;

   Data telah ditulis ke database AnalyticDB for PostgreSQL. Gambar berikut menunjukkan hasil yang dikembalikan.

   

Langkah selanjutnya

• Konektor AnalyticDB for PostgreSQL — referensi lengkap konektor, termasuk semua parameter dan pemetaan tipe data

• Konektor Datagen — referensi untuk sumber datagen yang digunakan dalam contoh ini

Referensi

• Datagen