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

CREATE TEMPORARY TABLE adb_table (
  `id` INT,
  `num` BIGINT,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'adb3.0',
  'url' = 'jdbc:mysql://<endpoint>:<port>/<databaseName>',
  'userName' = '<yourUsername>',
  'password' = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

Clé primaire et mode d'écriture

Lorsqu'une clé primaire est définie dans le DDL, le connecteur fonctionne en mode upsert : les clés primaires en double déclenchent une mise à jour plutôt qu'une nouvelle insertion. Le comportement SQL exact dépend du paramètre replaceMode :

• replace — utilise REPLACE INTO. Une clé primaire en double écrase la ligne existante dans son intégralité.

• upsert — utilise INSERT INTO ... ON DUPLICATE KEY UPDATE. Seuls les champs spécifiés sont mis à jour ; les autres champs conservent leurs valeurs actuelles.

• insert — utilise INSERT IGNORE INTO. Les clés primaires en double sont ignorées silencieusement ; la ligne existante est conservée.

En l'absence de clé primaire, le connecteur utilise toujours INSERT IGNORE INTO.

Le paramètre replaceMode nécessite AnalyticDB for MySQL V3.1.3.5 ou une version ultérieure et VVR 11.2 ou une version ultérieure pour les valeurs de chaîne ( replace , upsert , insert ). Les versions antérieures de VVR utilisent true (équivalent à replace ) et false (équivalent à upsert ). VVR 11.2 et les versions ultérieures restent compatibles avec true et false .

Mise en mémoire tampon des écritures

Le connecteur met les enregistrements en mémoire tampon et les vide par lots. Un vidage est déclenché lorsque l'une des conditions suivantes est remplie :

• Le nombre d'enregistrements en mémoire tampon atteint batchSize (par défaut : 1 000) ou bufferSize (par défaut : 1 000)

• Le temps écoulé depuis le dernier vidage atteint flushIntervalMs (par défaut : 3 000 ms)

Les paramètres batchSize et bufferSize ne prennent effet que lorsqu'une clé primaire est définie.

Mise en cache des tables de dimension

Le connecteur prend en charge trois stratégies de mise en cache pour les tables de dimension. La mise en cache réduit le nombre de recherches dans la table physique, mais nécessite davantage de mémoire.

Paramètres communs

Paramètres de la table de destination

Paramètres de la table de dimension

Table de destination

L'exemple suivant lit à partir d'une source datagen et écrit les données dans une table de destination 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;

Table de dimension

L'exemple suivant joint une source datagen à une table de dimension AnalyticDB for MySQL à l'aide d'une jointure temporelle. Les résultats sont écrits dans une destination 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;