Realtime Compute for Apache Flink:ApsaraDB RDS for MySQL

Perilaku penulisan sink

Setiap baris output dikonversi menjadi pernyataan SQL sebelum ditulis ke tabel sink:

• Tanpa primary key — menjalankan INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...);

• Dengan primary key — menjalankan INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...) ON DUPLICATE KEY UPDATE col1 = VALUES(col1), col2 = VALUES(col2), ...;

Konflik indeks unik: Jika tabel fisik memiliki batasan indeks unik selain primary key, penyisipan dua baris dengan primary key berbeda tetapi nilai indeks unik yang sama menyebabkan baris sebelumnya ditimpa, sehingga mengakibatkan kehilangan data.

Primary key auto-increment: Jangan deklarasikan kolom auto-increment dalam DDL Flink. Database secara otomatis menetapkan nilai-nilai tersebut. Konektor dapat menulis dan menghapus baris yang memiliki kolom auto-increment, tetapi tidak dapat memperbaruinya.

Kebijakan cache tabel dimensi

Konektor mendukung tiga kebijakan cache untuk pencarian tabel dimensi:

Tabel sink

CREATE TABLE rds_sink (
  id  INT,
  num BIGINT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'  = 'rds',
  'tableName'  = '<your-table-name>',
  'userName'   = '<your-user-name>',
  'password'   = '<your-password>',
  'url'        = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>?rewriteBatchedStatements=true'
);

Tabel dimensi

CREATE TABLE rds_dim (
  id1 INT,
  id2 VARCHAR
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>',
  'cache'     = 'NONE'
);

Parameter umum

Parameter tabel sink

Parameter tabel dimensi

Contoh tabel sink

Contoh berikut membaca dari sumber DataGen dan menulis ke tabel sink ApsaraDB RDS for MySQL.

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

CREATE TEMPORARY TABLE rds_sink (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>?rewriteBatchedStatements=true'
);

INSERT INTO rds_sink
SELECT * FROM datagen_source;

Contoh tabel dimensi

Contoh berikut melakukan join aliran terhadap tabel dimensi ApsaraDB RDS for MySQL menggunakan temporal join.

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

CREATE TEMPORARY TABLE rds_dim (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>'
);

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

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