Realtime Compute for Apache Flink:JDBC connector

General

Source options

Sink options

Dimension table options

PostgreSQL options

Partitioned scan

Untuk mengaktifkan pembacaan paralel dari tabel sumber, konfigurasikan opsi scan.partition.column, scan.partition.lower-bound, dan scan.partition.upper-bound bersama dengan scan.partition.num. Kolom partisi harus bertipe numeric atau TIMESTAMP. Untuk detail cara split dihitung, lihat Partitioned Scan dalam dokumentasi Apache Flink.

Lookup cache

Secara default, setiap kueri join lookup langsung mengakses database. Aktifkan cache LRU dengan mengatur kedua opsi lookup.cache.max-rows dan lookup.cache.ttl untuk mengurangi beban database dengan sedikit kompromi pada kesegaran data.

Idempotent writes

Saat tabel sink memiliki primary key, konektor mengeluarkan pernyataan upsert spesifik database. Untuk MySQL, konektor menggunakan INSERT ... ON DUPLICATE KEY UPDATE ....

Read from a database (source table)

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;

Write to a database (sink table)

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;

Enrich a stream with database lookups (dimension table)

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;