Realtime Compute for Apache Flink:AnalyticDB for MySQL V3.0

Kunci primary dan mode penulisan

Jika kunci primary didefinisikan dalam DDL, konektor beroperasi dalam mode upsert, di mana kunci primary duplikat akan memicu pembaruan alih-alih penyisipan baru. Perilaku SQL yang tepat bergantung pada parameter replaceMode:

• replace — menggunakan REPLACE INTO. Kunci primary duplikat akan menimpa seluruh baris yang sudah ada.

• upsert — menggunakan INSERT INTO ... ON DUPLICATE KEY UPDATE. Hanya bidang yang ditentukan yang diperbarui; bidang lain tetap mempertahankan nilai saat ini.

• insert — menggunakan INSERT IGNORE INTO. Kunci primary duplikat diabaikan secara diam-diam; baris yang sudah ada tetap dipertahankan.

Jika tidak ada kunci primary yang didefinisikan, konektor selalu menggunakan INSERT IGNORE INTO.

replaceMode memerlukan AnalyticDB for MySQL V3.1.3.5 atau versi yang lebih baru serta VVR 11.2 atau versi yang lebih baru untuk nilai string (replace, upsert, insert). Versi VVR sebelumnya menggunakan true (setara dengan replace) dan false (setara dengan upsert). VVR 11.2 dan versi setelahnya tetap kompatibel dengan true dan false.

Pembuferan penulisan

Konektor menyimpan catatan dalam memori dan mengirimkannya secara batch. Pengiriman dilakukan ketika salah satu kondisi berikut terpenuhi:

• Jumlah catatan yang dibuffer mencapai batchSize (default: 1.000) atau bufferSize (default: 1.000).

• Waktu sejak pengiriman terakhir mencapai flushIntervalMs (default: 3.000 ms).

Baik batchSize maupun bufferSize hanya berlaku jika kunci primary didefinisikan.

Peng-cache-an tabel dimensi

Konektor mendukung tiga kebijakan cache untuk tabel dimensi. Penggunaan cache mengurangi jumlah lookup ke tabel fisik tetapi memerlukan memori tambahan.

Parameter umum

Parameter tabel sink

Parameter tabel dimensi

Tabel sink

Contoh berikut membaca dari sumber datagen dan menulis data ke tabel sink AnalyticDB for MySQL.

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

CREATE TEMPORARY TABLE adb_sink (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector' = 'adb3.0',
  'url'       = 'jdbc:mysql://<endpoint>:<port>/<databaseName>',
  'userName'  = '<yourUsername>',
  'password'  = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

INSERT INTO adb_sink
SELECT * FROM datagen_source;

Tabel dimensi

Contoh berikut melakukan join antara sumber datagen dengan tabel dimensi AnalyticDB for MySQL menggunakan temporal join. Hasilnya ditulis ke sink blackhole.

CREATE TEMPORARY TABLE datagen_source (
  `a`       INT,
  `b`       VARCHAR,
  `c`       STRING,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE adb_dim (
  `a` INT,
  `b` VARCHAR,
  `c` VARCHAR
) WITH (
  'connector' = 'adb3.0',
  'url'       = 'jdbc:mysql://<endpoint>:<port>/<databaseName>',
  'userName'  = '<yourUsername>',
  'password'  = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  `a` INT,
  `b` VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT T.a, H.b
FROM datagen_source AS T
JOIN adb_dim FOR SYSTEM_TIME AS OF T.proctime AS H
  ON T.a = H.a;