Realtime Compute for Apache Flink:Connecteur JDBC

Général

Options de source

Options de puits (sink)

Options de table de dimension

Options PostgreSQL

Analyse partitionnée

Pour activer les lectures parallèles depuis une table source, configurez conjointement les options scan.partition.column, scan.partition.lower-bound, scan.partition.upper-bound et scan.partition.num. La colonne de partition doit être de type numérique ou TIMESTAMP. Pour plus de détails sur le calcul des splits, consultez la section Partitioned Scan de la documentation Apache Flink.

Cache de recherche

Par défaut, chaque requête de jointure de recherche interroge directement la base de données. Activez le cache LRU en définissant à la fois lookup.cache.max-rows et lookup.cache.ttl afin de réduire la charge sur la base de données, au prix de données légèrement obsolètes.

Écritures idempotentes

Lorsque la table puits possède une clé primaire, le connecteur émet des instructions upsert spécifiques à la base de données. Pour MySQL, le connecteur utilise l'instruction INSERT ... ON DUPLICATE KEY UPDATE ....

Lire depuis une base de données (table source)

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;

Écrire dans une base de données (table puits)

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;

Enrichir un flux avec des recherches en base de données (table de dimension)

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;