Realtime Compute for Apache Flink:AnalyticDB for MySQL V3.0

プライマリキーと書き込みモード

DDL でプライマリキーが定義されている場合、コネクタはアップサートモードで動作します。このモードでは、重複するプライマリキーは新しい挿入ではなく更新をトリガーします。具体的な SQL の動作は replaceMode パラメーターによって決まります：

• replace — REPLACE INTO を使用します。プライマリキーが重複する場合、既存の行全体が上書きされます。

• upsert — INSERT INTO ... ON DUPLICATE KEY UPDATE を使用します。指定されたフィールドのみが更新され、他のフィールドは現在の値を保持します。

• insert — INSERT IGNORE INTO を使用します。重複するプライマリキーは警告なしに無視され、既存の行が保持されます。

プライマリキーが定義されていない場合、コネクタは常に INSERT IGNORE INTO を使用します。

replaceMode は、文字列値 (replace、upsert、insert) を使用する場合、AnalyticDB for MySQL V3.1.3.5 以降および VVR 11.2 以降が必要です。以前の VVR バージョンでは、true (replace に相当) と false (upsert に相当) を使用します。VVR 11.2 以降は、true および false との互換性を維持しています。

書き込みバッファリング

コネクタはレコードをメモリにバッファリングし、バッチでフラッシュします。フラッシュは、以下のいずれかの条件が満たされたときにトリガーされます：

• バッファリングされたレコード数が batchSize (デフォルト：1,000) または bufferSize (デフォルト：1,000) に達したとき

• 最後のフラッシュからの時間が flushIntervalMs (デフォルト：3,000 ms) に達したとき

batchSize と bufferSize は、プライマリキーが定義されている場合にのみ有効です。

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

コネクタは、ディメンションテーブルに対して 3 つのキャッシュポリシーをサポートしています。キャッシュを使用すると、物理テーブルに対するルックアップが減少しますが、追加のメモリが必要になります。

共通パラメーター

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

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

結果テーブル

次の例では、datagen ソースからデータを読み取り、AnalyticDB for MySQL の結果テーブルに書き込みます。

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

CREATE TEMPORARY TABLE adb_sink (
  `name` VARCHAR,
  `age`  INT
) WITH (
  'connector' = 'adb3.0',
  'url'       = 'jdbc:mysql://<endpoint>:<port>/<databaseName>',
  'userName'  = '<yourUsername>',
  'password'  = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

INSERT INTO adb_sink
SELECT * FROM datagen_source;

ディメンションテーブル

次の例では、datagen ソースと AnalyticDB for MySQL のディメンションテーブルをテンポラル結合で結合します。結果は blackhole の結果テーブルに書き込まれます。

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

CREATE TEMPORARY TABLE adb_dim (
  `a` INT,
  `b` VARCHAR,
  `c` VARCHAR
) WITH (
  'connector' = 'adb3.0',
  'url'       = 'jdbc:mysql://<endpoint>:<port>/<databaseName>',
  'userName'  = '<yourUsername>',
  'password'  = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

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

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