Realtime Compute for Apache Flink:Create a MySQL dimension table

What is a MySQL dimension table?

MySQL dimension tables support all databases that are compatible with the MySQL protocol. The databases include ApsaraDB RDS for MySQL, PolarDB for MySQL, and self-managed MySQL databases. ApsaraDB RDS for MySQL dimension tables are customized for ApsaraDB RDS for MySQL databases and support more parameters.

Prerequisites

• A MySQL database and a MySQL table are created. For more information, see Create a database and account for an ApsaraDB RDS for MySQL instance, Create a database and account for a PolarDB for MySQL instance, or Create a database and account for a self-managed MySQL instance.
• An IP address whitelist is configured. For more information, see Configure an ApsaraDB RDS for MySQL whitelist, Configure a PolarDB for MySQL whitelist, or Configure a self-managed MySQL whitelist.

Limits

Only Flink that uses Ververica Runtime (VVR) 4.0.11 or later supports MySQL connectors.

DDL syntax

CREATE TABLE mysql_dim (
   id INT,
   name STRING,
   PRIMARY KEY (id) NOT ENFORCED
) WITH (
   'connector' = 'mysql',
   'hostname' = '<yourHostname>',
   'port' = '3306',
   'username' = '<yourUsername>',
   'password' = '<yourPassword>',
   'database-name' = '<yourDatabaseName>',
   'table-name' = '<yourTableName>'
);

Parameters in the WITH clause

Data type mappings

Sample code

CREATE TEMPORARY TABLE datagen_source(
  a INT,
  b BIGINT,
  c STRING,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE mysql_dim (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

CREATE TEMPORARY TABLE blackhole_sink(
  a INT,
  b STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT T.a, H.b
FROM datagen_source AS T JOIN mysql_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H ON T.a = H.a;