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

Primary key and write mode

When a primary key is defined in the DDL, the connector operates in upsert mode, where duplicate primary keys trigger an update rather than a new insert. The exact SQL behavior depends on the replaceMode parameter:

• replace — uses REPLACE INTO. A duplicate primary key overwrites the entire existing row.

• upsert — uses INSERT INTO ... ON DUPLICATE KEY UPDATE. Only the specified fields are updated; other fields retain their current values.

• insert — uses INSERT IGNORE INTO. Duplicate primary keys are silently ignored; the existing row is kept.

When no primary key is defined, the connector always uses INSERT IGNORE INTO.

replaceMode requires AnalyticDB for MySQL V3.1.3.5 or later and VVR 11.2 or later for the string values (replace, upsert, insert). Earlier VVR versions use true (equivalent to replace) and false (equivalent to upsert). VVR 11.2 and later remain compatible with true and false.

Write buffering

The connector buffers records in memory and flushes them in batches. A flush is triggered when either of the following conditions is met:

• The number of buffered records reaches batchSize (default: 1,000) or bufferSize (default: 1,000)

• The time since the last flush reaches flushIntervalMs (default: 3,000 ms)

Both batchSize and bufferSize take effect only when a primary key is defined.

Dimension table caching

The connector supports three cache policies for dimension tables. Caching reduces lookups against the physical table but requires extra memory.

Common parameters

Sink table parameters

Dimension table parameters

Sink table

The following example reads from a datagen source and writes the data to an AnalyticDB for MySQL sink table.

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;

Dimension table

The following example joins a datagen source with an AnalyticDB for MySQL dimension table using a temporal join. Results are written to a blackhole sink.

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;