Realtime Compute for Apache Flink:Connecteur JDBC

Prérequis

Avant de commencer, assurez-vous que :

• La base de données et la table cibles existent déjà.

• Le fichier JAR du pilote JDBC correspondant à votre base de données est disponible pour téléchargement.

Limites

• Lectures bornées : Une table source JDBC constitue une source bornée. La tâche se termine une fois toutes les lignes lues. Pour capturer les données de modification en temps réel, utilisez plutôt un connecteur Change Data Capture (CDC). Consultez les rubriques Créer une table source MySQL CDC et Créer une table source PostgreSQL CDC (aperçu public).

• Version de PostgreSQL : L'écriture dans PostgreSQL nécessite la version 9.5 ou ultérieure, car le mécanisme de sortie (sink) s'appuie sur la clause ON CONFLICT.

• Téléchargement du pilote JDBC : Téléchargez le fichier JAR du pilote JDBC en tant que dépendance avant d'exécuter votre job. Pilotes courants :

  Base de données	Group ID	Artifact ID
  MySQL	mysql	mysql-connector-java
  Oracle	com.oracle.database.jdbc	ojdbc8
  PostgreSQL	org.postgresql	postgresql

  Pour les pilotes non répertoriés ici, vérifiez la compatibilité avant utilisation.

• Comportement upsert MySQL : Lors de l'écriture dans une table de sortie MySQL dotée d'une clé primaire, le connecteur émet des instructions INSERT INTO ... ON DUPLICATE KEY UPDATE ....

  Avertissement

  L'insertion de lignes comportant des valeurs d'index unique dupliquées, même si les clés primaires diffèrent, entraîne l'écrasement des lignes existantes dans toute table physique soumise à une contrainte d'index unique, ce qui provoque une perte de données.

Créer une table JDBC

CREATE TABLE jdbc_table (
  `id`   BIGINT,
  `name` VARCHAR,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'  = 'jdbc',
  'url'        = 'jdbc:<db-type>://<host>:<port>/<database>',
  'table-name' = '<yourTable>',
  'username'   = '<yourUsername>',
  'password'   = '<yourPassword>'
);

Options du connecteur

Comportements clés

Exemples

Les trois exemples suivants utilisent un connecteur blackhole ou datagen comme contrepartie légère, permettant ainsi aux instructions d'être autonomes.

CREATE TEMPORARY TABLE jdbc_source (
  `id`   INT,
  `name` VARCHAR
) WITH (
  'connector'  = 'jdbc',
  'url'        = 'jdbc:mysql://localhost:3306/mydb',
  'table-name' = '<yourTable>',
  'username'   = '<yourUsername>',
  'password'   = '<yourPassword>'
);

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

INSERT INTO blackhole_sink SELECT * FROM jdbc_source;

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

CREATE TEMPORARY TABLE jdbc_sink (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector'  = 'jdbc',
  'url'        = 'jdbc:mysql://localhost:3306/mydb',
  'table-name' = '<yourTable>',
  'username'   = '<yourUsername>',
  'password'   = '<yourPassword>'
);

INSERT INTO jdbc_sink SELECT * FROM datagen_source;

CREATE TEMPORARY TABLE datagen_source (
  `id`       INT,
  `data`     BIGINT,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE jdbc_dim (
  `id`   INT,
  `name` VARCHAR
) WITH (
  'connector'  = 'jdbc',
  'url'        = 'jdbc:mysql://localhost:3306/mydb',
  'table-name' = '<yourTable>',
  'username'   = '<yourUsername>',
  'password'   = '<yourPassword>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  `id`   INT,
  `data` BIGINT,
  `name` VARCHAR
) WITH (
  'connector' = 'blackhole'
);

-- Look up the matching name for each stream record at processing time
INSERT INTO blackhole_sink
SELECT T.`id`, T.`data`, H.`name`
FROM datagen_source AS T
JOIN jdbc_dim FOR SYSTEM_TIME AS OF T.proctime AS H
  ON T.id = H.id;

Mappings de types de données

Étapes suivantes

• Créer une table source MySQL CDC : capturez les données de modification en temps réel depuis MySQL.

• Créer une table source PostgreSQL CDC (aperçu public) : capturez les données de modification en temps réel depuis PostgreSQL.