Menulis data ke AnalyticDB for PostgreSQL

Topik ini menjelaskan cara menulis data ke AnalyticDB for PostgreSQL menggunakan Realtime Compute for Apache Flink.

Batasan penggunaan

Fitur ini tidak mendukung mode Serverless AnalyticDB for PostgreSQL.
Hanya Realtime Compute for Apache Flink VVR 6.0.0 atau yang lebih baru yang mendukung konektor AnalyticDB for PostgreSQL.
Hanya Realtime Compute for Apache Flink VVR 8.0.1 atau yang lebih baru yang mendukung AnalyticDB for PostgreSQL V7.0.

Catatan
Jika Anda menggunakan konektor kustom, lihat Mengelola konektor kustom untuk informasi selengkapnya.

Prasyarat

Ruang kerja Fully Managed Flink telah dibuat. Untuk informasi selengkapnya, lihat Aktifkan Realtime Compute for Apache Flink.
Instans AnalyticDB for PostgreSQL telah dibuat. Untuk informasi selengkapnya, lihat Buat instans.
Instans AnalyticDB for PostgreSQL dan ruang kerja Fully Managed Flink harus berada dalam VPC yang sama.

Konfigurasi instans AnalyticDB for PostgreSQL

Masuk ke Konsol AnalyticDB for PostgreSQL.
Tambahkan Blok CIDR ruang kerja Flink ke daftar putih instans AnalyticDB for PostgreSQL. Untuk informasi selengkapnya tentang cara mengonfigurasi daftar putih, lihat Konfigurasi daftar putih.
Klik Log On to Database. Untuk informasi selengkapnya tentang cara lain menghubungkan ke database, lihat Hubungkan ke instans menggunakan client.
Buat tabel pada instans AnalyticDB for PostgreSQL.
Pernyataan SQL contoh untuk membuat tabel:
```
CREATE TABLE test_adbpg_table(
b1 int,
b2 int,
b3 text,
PRIMARY KEY(b1)
);
```

Konfigurasi Realtime Compute for Flink

Masuk ke Konsol Realtime Compute.
Pada tab Fully Managed Flink, klik Console di kolom Actions ruang kerja target.
Di panel navigasi sebelah kiri, klik Connectors.
Pada halaman Connectors, klik Create Custom Connector.
Unggah file JAR konektor kustom.
Catatan
- Dapatkan paket JAR konektor Flink AnalyticDB for PostgreSQL kustom. Untuk informasi selengkapnya, lihat AnalyticDB PostgreSQL Connector.
- Versi paket JAR harus sesuai dengan versi engine Flink platform Realtime Compute.
Setelah unggah selesai, klik Next.

Sistem akan mengurai konektor kustom yang diunggah. Jika penguraian berhasil, Anda dapat melanjutkan ke langkah berikutnya. Jika penguraian gagal, periksa apakah kode konektor kustom yang diunggah memenuhi standar komunitas Apache Flink.
Klik Finish.

Konektor kustom yang dibuat akan muncul dalam daftar konektor.

Buat pekerjaan Flink

Masuk ke Konsol Realtime Compute. Pada tab Fully Managed Flink, klik Console di kolom Actions ruang kerja target.
Di panel navigasi sebelah kiri, klik SQL Development. Klik New, pilih Blank Streaming Job Draft, lalu klik Next.

Pada kotak dialog New File Draft, konfigurasikan parameter pekerjaan.

Job Parameter	Description	Example
Name	Nama pekerjaan. Catatan Nama pekerjaan harus unik dalam proyek saat ini.	adbpg-test
Location	Folder tempat file kode pekerjaan disimpan. Anda juga dapat mengklik ikon di sisi kanan folder yang sudah ada untuk membuat subfolder.	Job Drafts
Engine Version	Versi engine Flink yang digunakan oleh pekerjaan ini. Untuk informasi selengkapnya tentang nomor versi, pemetaan versi, dan tonggak siklus hidup, lihat Engine versions.	vvr-6.0.7-flink-1.15

Klik Create.

Tulis kode pekerjaan.

Buat tabel sumber acak datagen_source dan tabel tujuan test_adbpg_table di AnalyticDB for PostgreSQL. Salin kode pekerjaan berikut ke editor teks pekerjaan.

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'
);
CREATE TABLE test_adbpg_table (
    `B1` bigint   ,
    `B2` bigint  ,
    `B3` VARCHAR ,
    `B4` VARCHAR,
     PRIMARY KEY(B1) not ENFORCED
) with (
   'connector' = 'adbpg-nightly-1.13',
   'password' = 'xxx',
   'tablename' = 'test_adbpg_table',
   'username' = 'xxxx',
   'url' = 'jdbc:postgresql://url:5432/schema',
   'maxretrytimes' = '2',
   'batchsize' = '50000',
   'connectionmaxactive' = '5',
   'conflictmode' = 'ignore',
   'usecopy' = '0',
   'targetschema' = 'public',
   'exceptionmode' = 'ignore',
   'casesensitive' = '0',
   'writemode' = '1',
   'retrywaittime' = '200'
);

Anda tidak perlu mengubah parameter tabel datagen_source. Anda harus mengubah parameter tabel test_adbpg_table sesuai dengan kebutuhan bisnis aktual Anda. Tabel berikut menjelaskan parameter-parameter tersebut.

Parameter	Required	Description
connector	Yes	Nama konektor. Atur parameter ini menjadi `adbpg-nightly-<version number>`, misalnya `adbpg-nightly-1.13`.
url	Yes	URL JDBC instans AnalyticDB for PostgreSQL. Format: `jdbc:postgresql://<internal endpoint>:<port>/<database name>`. Contoh: `jdbc:postgresql://gp-xxxxxx.gpdb.cn-chengdu.rds.aliyuncs.com:5432/postgres`.
tablename	Yes	Nama tabel AnalyticDB for PostgreSQL.
username	Yes	Akun database instans AnalyticDB for PostgreSQL.
password	Yes	Kata sandi akun database untuk instans AnalyticDB for PostgreSQL.
maxretrytimes	No	Jumlah maksimum percobaan ulang setelah kegagalan eksekusi SQL. Nilai default: 3.
batchsize	No	Jumlah maksimum catatan data yang ditulis dalam satu batch. Nilai default: 50000.
exceptionmode	No	Kebijakan penanganan kesalahan ketika terjadi exception selama penulisan data. Nilai yang valid: ignore: mengabaikan data yang menyebabkan exception. Ini adalah nilai default. strict: memicu failover dan melaporkan kesalahan ketika terjadi exception penulisan data.
conflictmode	No	Kebijakan penanganan konflik kunci primer atau indeks unik. Nilai yang valid: ignore: mengabaikan konflik kunci primer dan mempertahankan data yang ada. strict: memicu failover dan melaporkan kesalahan ketika terjadi konflik kunci primer. update: memperbarui data ketika terjadi konflik kunci primer. upsert: menulis data menggunakan metode UPSERT ketika terjadi konflik kunci primer. Ini adalah nilai default. AnalyticDB for PostgreSQL mengimplementasikan UPSERT menggunakan INSERT ON CONFLICT dan COPY ON CONFLICT. Jika tabel tujuan adalah tabel partisi, versi kernel minor harus V6.3.6.1 atau yang lebih baru. Untuk informasi selengkapnya tentang cara meningkatkan versi kernel minor, lihat Upgrade the engine version.
targetschema	No	Skema database AnalyticDB for PostgreSQL. Nilai default: public.
writemode	No	Mode penulisan data. Nilai yang valid: 0: menulis data menggunakan BATCH INSERT. 1: menulis data menggunakan API COPY. Ini adalah nilai default. 2: menulis data menggunakan BATCH UPSERT.
verbose	No	Menentukan apakah log waktu proses konektor akan dikeluarkan. Nilai yang valid: 0: tidak mengeluarkan log waktu proses. Ini adalah nilai default. 1: mengeluarkan log waktu proses.
retrywaittime	No	Interval antar percobaan ulang ketika terjadi exception. Satuan: milidetik. Nilai default: 100.
batchwritetimeoutms	No	Waktu akumulasi batch maksimum untuk penulisan batch. Ketika waktu ini terlampaui, batch yang terakumulasi akan ditulis. Satuan: milidetik. Nilai default: 50000.
connectionmaxactive	No	Parameter kolam koneksi. Parameter ini menentukan jumlah maksimum koneksi bersamaan dalam kolam koneksi satu Task Manager. Nilai default: 5.
casesensitive	No	Menentukan apakah nama kolom dan nama tabel peka huruf besar/kecil. Nilai yang valid: 0: tidak peka huruf besar/kecil. Ini adalah nilai default. 1: peka huruf besar/kecil.

Catatan

Pemetaan parameter dan tipe didukung. Untuk informasi selengkapnya, lihat dokumentasi konektor untuk AnalyticDB for PostgreSQL (ADB PG).

Jalankan pekerjaan.
1. Di bagian atas halaman pengembangan pekerjaan, klik Deploy. Di kotak dialog yang muncul, klik OK.
  
  Catatan
  Kluster Session cocok untuk pengembangan dan pengujian di lingkungan non-produksi. Anda dapat menggunakan Kluster Session untuk men-debug pekerjaan, meningkatkan pemanfaatan resource Job Manager (JM), dan mempercepat startup pekerjaan. Namun, kami tidak merekomendasikan Anda mengirimkan pekerjaan ke Kluster Session karena pertimbangan stabilitas bisnis. Untuk informasi selengkapnya, lihat Debug a job.
2. Pada halaman Deployments, klik Resume di kolom Actions pekerjaan target.
3. Klik Resume.

Verifikasi hasil

Hubungkan ke database AnalyticDB for PostgreSQL. Untuk informasi selengkapnya, lihat Hubungkan ke instans menggunakan client.
Eksekusi pernyataan berikut untuk mengkueri tabel test_adbpg_table.
```
SELECT * FROM test_adbpg_table;
```
Data telah ditulis ke AnalyticDB for PostgreSQL sesuai harapan. Gambar berikut menunjukkan contoh respons.

Hasil kueri berisi tiga kolom: b1 (int4), b2 (int4), dan b3 (text). Beberapa baris data dikembalikan, yang mengonfirmasi bahwa data telah disinkronkan ke tabel tujuan.

AnalyticDB:Menulis data ke AnalyticDB for PostgreSQL