Realtime Compute for Apache Flink:ApsaraDB RDS for MySQL

結果テーブルの書き込み動作

各出力行は、結果テーブルに書き込まれる前に SQL ステートメントに変換されます。

• プライマリキーなし — INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...); を実行します。

• プライマリキーあり — INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...) ON DUPLICATE KEY UPDATE col1 = VALUES(col1), col2 = VALUES(col2), ...; を実行します。

一意なインデックスの競合： 物理テーブルにプライマリキーに加えて一意なインデックス制約がある場合、異なるプライマリキーで同じ一意なインデックス値を持つ 2 つの行を挿入すると、先の行が上書きされ、データ損失が発生します。

自動インクリメントのプライマリキー： Flink DDL で自動インクリメントフィールドを宣言しないでください。データベースがこれらの値を自動的に割り当てます。コネクタは自動インクリメントフィールドを持つ行の書き込みと削除はできますが、更新はできません。

ディメンションテーブルのキャッシュポリシー

コネクタは、ディメンションテーブルのルックアップに対して 3 つのキャッシュポリシーをサポートしています。

結果テーブル

CREATE TABLE rds_sink (
  id  INT,
  num BIGINT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'  = 'rds',
  'tableName'  = '<your-table-name>',
  'userName'   = '<your-user-name>',
  'password'   = '<your-password>',
  'url'        = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>?rewriteBatchedStatements=true'
);

ディメンションテーブル

CREATE TABLE rds_dim (
  id1 INT,
  id2 VARCHAR
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>',
  'cache'     = 'NONE'
);

共通パラメーター

シンクテーブル パラメーター

ディメンションテーブルのパラメーター

結果テーブルの例

次の例では、DataGen ソースから読み取り、ApsaraDB RDS for MySQL の結果テーブルに書き込みます。

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

CREATE TEMPORARY TABLE rds_sink (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>?rewriteBatchedStatements=true'
);

INSERT INTO rds_sink
SELECT * FROM datagen_source;

ディメンションテーブルの例

次の例では、テンポラル結合を使用して、ストリームを ApsaraDB RDS for MySQL のディメンションテーブルに対して結合します。

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

CREATE TEMPORARY TABLE rds_dim (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'rds',
  'tableName' = '<your-table-name>',
  'userName'  = '<your-user-name>',
  'password'  = '<your-password>',
  'url'       = 'jdbc:mysql://<internal-endpoint>:<port>/<database-name>'
);

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 rds_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H
ON T.a = H.a;