Realtime Compute for Apache Flink:MySQL SQL コネクタ

SQL

MySQL コネクタは、SQL ジョブでソーステーブル、ディメンションテーブル、または結果テーブルとして使用できます。

構文

CREATE TEMPORARY TABLE mysqlcdc_source (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

WITH パラメーター

• 全般

  パラメーター	説明	必須	データの型	デフォルト値	注釈
  connector	テーブルのタイプ。	はい	STRING	なし	ソーステーブルとして使用する場合、このパラメーターを mysql-cdc または mysql に設定できます。これらは同等です。ディメンションテーブルまたは結果テーブルとして使用する場合、値は mysql である必要があります。
  hostname	MySQL データベースの IP アドレスまたはホスト名。	はい	STRING	なし	VPC (Virtual Private Cloud) アドレスを指定することを推奨します。 説明 MySQL データベースと Realtime Compute for Apache Flink が同じ VPC にない場合、VPC 間のネットワーク接続を確立するか、パブリックエンドポイントを使用してデータベースにアクセスする必要があります。詳細については、「ワークスペースの管理と運用」および「Flink 完全管理クラスターがインターネットにアクセスする方法」をご参照ください。
  username	MySQL データベースサービスのユーザー名。	はい	STRING	なし	なし。
  password	MySQL データベースサービスのパスワード。	はい	STRING	なし	なし。
  database-name	MySQL データベースの名前。	はい	STRING	なし	データベースをソーステーブルとして使用する場合、データベース名に正規表現を使用して複数のデータベースからデータを読み取ることができます。 正規表現を使用する場合、文字列の先頭と末尾を一致させるために ^ および $ 記号を使用しないでください。詳細については、table-name パラメーターの注釈をご参照ください。
  table-name	MySQL テーブルの名前。	はい	STRING	なし	ソーステーブル名に正規表現を使用して、複数のテーブルからデータを読み取ることができます。 正規表現を使用する場合、文字列の先頭と末尾を一致させるために ^ および $ 記号を使用しないでください。詳細については、以下の注釈をご参照ください。 説明 MySQL CDC ソーステーブルが正規表現を使用してテーブル名を照合する場合、指定した database-name と table-name を文字列 \\. で連結して、完全パスの正規表現を形成します。VVR 8.0.1 より前では、文字 . が使用されていました。コネクタは、この正規表現を使用して MySQL データベース内のテーブルの完全修飾名を照合します。 たとえば、'database-name'='db_.*' および 'table-name'='tb_.+' と設定した場合、コネクタは正規表現 db_.*\\.tb_.+ を使用して完全修飾テーブル名を照合し、どのテーブルを読み取るかを決定します。VVR 8.0.1 より前では、正規表現は db_.*.tb_.+ でした。
  port	MySQL データベースサービスのポート番号。	いいえ	INTEGER	3306	なし。

• ソーステーブルのみ

  パラメーター	説明	必須	データの型	デフォルト値	注釈
  server-id	データベースクライアントの数値 ID。	いいえ	STRING	5400 から 6400 の間のランダムな値が生成されます。	この ID は MySQL クラスター内でグローバルに一意である必要があります。同じデータベースに接続する各ジョブには異なる ID を設定してください。 このパラメーターは、5400-5408 のような ID 範囲形式もサポートします。増分読み取りが有効な場合、同時読み取りがサポートされます。この場合、各同時リーダーが異なる ID を使用するように ID 範囲を設定します。詳細については、「サーバー ID の使用」をご参照ください。
  scan.incremental.snapshot.enabled	増分スナップショットを有効にするかどうかを指定します。	いいえ	BOOLEAN	true	増分スナップショットはデフォルトで有効になっています。増分スナップショットは、完全なデータスナップショットを読み取るための新しいメカニズムです。古いスナップショット読み取り方法と比較して、増分スナップショットには次のような多くの利点があります： ソースは完全なデータを並行して読み取ることができます。 ソースは完全なデータを読み取る際にチャンクレベルのチェックポイントをサポートします。 ソースは完全なデータを読み取る際にグローバル読み取りロック (FLUSH TABLES WITH read lock) を取得する必要がありません。 ソースが同時読み取りをサポートするようにしたい場合、各同時リーダーには一意のサーバー ID が必要です。したがって、server-id は 5400-6400 のような範囲でなければならず、範囲のサイズは同時実行数以上でなければなりません。 説明 この設定項目は Ververica Runtime (VVR) 11.1 以降では削除されています。
  scan.incremental.snapshot.chunk.size	各チャンクのサイズ (行数)。	いいえ	INTEGER	8096	増分スナップショット読み取りが有効な場合、テーブルは複数のチャンクに分割されて読み取られます。チャンクのデータは、完全に読み取られるまでメモリにキャッシュされます。 各チャンクに含まれる行数が少ないほど、テーブル内のチャンクの総数は多くなります。これにより障害復旧の粒度は小さくなりますが、メモリ不足 (OOM) エラーや全体的なスループットの低下につながる可能性があります。したがって、トレードオフを考慮し、適切なチャンクサイズを設定する必要があります。
  scan.snapshot.fetch.size	テーブルの完全データを読み取る際に一度にプルするレコードの最大数。	いいえ	INTEGER	1024	なし。
  scan.startup.mode	データ消費の起動モード。	いいえ	STRING	initial	有効な値： initial (デフォルト)：初回起動時またはステートレス起動時に、コネクタは完全な既存データをスキャンし、その後最新のバイナリログデータを読み取ります。 latest-offset：初回起動時またはステートレス起動時に、コネクタは既存データをスキャンしません。バイナリログの末尾から読み取りを開始します。つまり、コネクタの起動後に加えられた最新の変更のみを読み取ります。 earliest-offset：コネクタは既存データをスキャンしません。利用可能な最も古いバイナリログから読み取りを開始します。 specific-offset：コネクタは既存データをスキャンしません。特定のバイナリログオフセットから開始します。scan.startup.specific-offset.file と scan.startup.specific-offset.pos の両方を設定するか、scan.startup.specific-offset.gtid-set のみを設定して特定の GTID セットから開始することでオフセットを指定できます。 timestamp：コネクタは既存データをスキャンしません。指定されたタイムスタンプからバイナリログの読み取りを開始します。タイムスタンプは scan.startup.timestamp-millis でミリ秒単位で指定します。 重要 earliest-offset、specific-offset、または timestamp 起動モードを使用する場合、指定されたバイナリログ消費位置とジョブ起動時間の間に対応するテーブルのスキーマが変更されないようにしてください。これにより、スキーマの不一致によるエラーを防ぎます。
  scan.startup.specific-offset.file	specific-offset 起動モードを使用する場合の開始オフセットのバイナリログファイル名。	いいえ	STRING	なし	このパラメーターを使用する場合、scan.startup.mode を specific-offset に設定する必要があります。ファイル名の例：mysql-bin.000003。
  scan.startup.specific-offset.pos	specific-offset 起動モードを使用する場合の、指定されたバイナリログファイル内の開始オフセット。	いいえ	INTEGER	なし	このパラメーターを使用する場合、scan.startup.mode を specific-offset に設定する必要があります。
  scan.startup.specific-offset.gtid-set	specific-offset 起動モードを使用する場合の開始オフセットの GTID セット。	いいえ	STRING	なし	このパラメーターを使用する場合、scan.startup.mode を specific-offset に設定する必要があります。GTID セットの例：24DA167-0C0C-11E8-8442-00059A3C7B00:1-19。
  scan.startup.timestamp-millis	timestamp 起動モードを使用する場合の開始オフセットのタイムスタンプ (ミリ秒)。	いいえ	LONG	なし	このパラメーターを使用する場合、scan.startup.mode を timestamp に設定する必要があります。タイムスタンプ単位はミリ秒です。 重要 時間を指定すると、MySQL CDC は各バイナリログファイルの初期イベントを読み取ってそのタイムスタンプを決定しようとします。その後、指定された時間に対応するバイナリログファイルを見つけます。指定されたタイムスタンプに対応するバイナリログファイルがデータベースからクリアされておらず、読み取り可能であることを確認してください。
  server-time-zone	データベースが使用するセッションタイムゾーン。	いいえ	STRING	このパラメーターを指定しない場合、システムは Flink ジョブのランタイムの環境タイムゾーンをデータベースサーバーのタイムゾーンとして使用します。これは選択したゾーンのタイムゾーンです。	例：Asia/Shanghai。このパラメーターは、MySQL の TIMESTAMP 型が STRING 型にどのように変換されるかを制御します。詳細については、「Debezium の時間値」をご参照ください。
  debezium.min.row.count.to.stream.results	テーブルの行数がこの値より大きい場合、バッチ読み取りモードが使用されます。	いいえ	INTEGER	1000	Flink は、次のいずれかの方法で MySQL ソーステーブルからデータを読み取ります： 完全読み取り：テーブル全体のデータを直接メモリに読み込みます。この方法は高速ですが、対応する量のメモリを消費します。ソーステーブルが非常に大きい場合、OOM エラーのリスクがあります。 バッチ読み取り：データを複数のバッチで読み取ります。バッチごとに特定の行数で、すべてのデータが読み取られるまで続きます。この方法は、大きなテーブルを読み取る際の OOM リスクを回避しますが、比較的低速です。
  connect.timeout	MySQL データベースサーバーへの接続がタイムアウトするまで待機する最大時間 (再試行前)。	いいえ	DURATION	30s	なし。
  connect.max-retries	MySQL データベースサービスへの接続に失敗した後の最大再試行回数。	いいえ	INTEGER	3	なし。
  connection.pool.size	データベース接続プールのサイズ。	いいえ	INTEGER	20	データベース接続プールは接続を再利用するために使用され、これによりデータベース接続の数を減らすことができます。
  jdbc.properties.*	JDBC URL のカスタム接続パラメーター。	いいえ	STRING	なし	カスタム接続パラメーターを渡すことができます。たとえば、SSL プロトコルを使用しないようにするには、'jdbc.properties.useSSL' = 'false' を設定します。 サポートされている接続パラメーターの詳細については、「MySQL 設定プロパティ」をご参照ください。
  debezium.*	バイナリログを読み取るための Debezium のカスタムパラメーター。	いいえ	STRING	なし	カスタム Debezium パラメーターを渡すことができます。たとえば、'debezium.event.deserialization.failure.handling.mode'='ignore' を使用して、解析エラーの処理ロジックを指定します。 警告 Debezium パラメーターを任意に変更しないでください。これにより、コネクタがデータを誤って読み取る可能性があります。たとえば、debezium.binlog.buffer.size パラメーターの設定は許可されていません。
  heartbeat.interval	ソースがハートビートイベントを使用してバイナリログオフセットを進める間隔。	いいえ	DURATION	30s	ハートビートイベントは、ソースのバイナリログオフセットを進めるために使用されます。これは、更新が頻繁でない MySQL のテーブルに非常に役立ちます。このようなテーブルでは、バイナリログオフセットは自動的に進みません。ハートビートイベントはバイナリログオフセットを前進させることができ、期限切れのバイナリログオフセットによって引き起こされる問題を回避します。期限切れのバイナリログオフセットは、ジョブの失敗と回復不能を引き起こし、ステートレスな再起動が必要になる可能性があります。
  scan.incremental.snapshot.chunk.key-column	スナップショットフェーズ中のシャーディングの分割列として使用する列を指定します。	「注釈」列をご参照ください。	STRING	なし	主キーのないテーブルでは必須です。選択された列は非ヌル型 (NOT NULL) である必要があります。 主キーのあるテーブルではオプションです。主キーから 1 つの列のみを選択できます。
  rds.region-id	Alibaba Cloud ApsaraDB RDS for MySQL インスタンスのリージョン ID。	OSS からのアーカイブログ読み取り機能を使用する場合に必須です。	STRING	なし	リージョン ID の詳細については、「リージョンとゾーン」をご参照ください。 重要 MySQL CDC の GTID 文字列はランダムに生成され、バイナリログファイルオフセットのように単調増加ではないため、ファイル内で GTID を見つけるには、OSS からすべてのアーカイブログをダウンロードして解析する必要があります。このプロセスは非常にリソースを消費し、時間がかかるため、GTID オフセットに依存する機能は実現不可能です。したがって、OSS アーカイブログ機能は、指定されたタイムスタンプまたは指定されたバイナリログファイルオフセットからの開始のみをサポートします。指定された GTID からの開始や、アーカイブログ内のプライマリ/セカンダリ スイッチオーバーのシナリオはサポートしていません。これは、MySQL のプライマリ/セカンダリ スイッチオーバーが GTID に依存するためです。使用前にこの機能を慎重に評価してください。
  rds.access-key-id	Alibaba Cloud ApsaraDB RDS for MySQL アカウントの AccessKey ID。	OSS からのアーカイブログ読み取り機能を使用する場合に必須です。	STRING	なし	詳細については、「AccessKey ID と AccessKey Secret の表示方法」をご参照ください。 重要 AccessKey 情報の漏洩を防ぐため、シークレット管理機能を使用して AccessKey ID を指定してください。詳細については、「変数の管理」をご参照ください。
  rds.access-key-secret	Alibaba Cloud ApsaraDB RDS for MySQL アカウントの AccessKey Secret。	OSS からのアーカイブログ読み取り機能を使用する場合に必須です。	STRING	なし	詳細については、「AccessKey ID と AccessKey Secret の表示方法」をご参照ください。 重要 AccessKey 情報の漏洩を防ぐため、シークレット管理機能を使用して AccessKey Secret を指定してください。詳細については、「変数の管理」をご参照ください。
  rds.db-instance-id	Alibaba Cloud ApsaraDB RDS for MySQL インスタンスの ID。	OSS からのアーカイブログ読み取り機能を使用する場合に必須です。	STRING	なし	なし。
  rds.main-db-id	Alibaba Cloud ApsaraDB RDS for MySQL インスタンスのプライマリデータベース番号。	いいえ	STRING	なし	プライマリデータベース番号の取得方法については、「ApsaraDB RDS for MySQL のログバックアップ」をご参照ください。 VVR 8.0.7 以降でのみサポートされています。 説明 このパラメーターが指定されていない場合、VVR 11.7 以降は ApsaraDB RDS for MySQL の接続情報に基づいてプライマリデータベース番号を自動的にクエリします。
  rds.download.timeout	OSS から単一のアーカイブログをダウンロードする際のタイムアウト期間。	いいえ	DURATION	60s	なし。
  rds.endpoint	OSS バイナリログ情報を取得するためのサービスエンドポイント。	いいえ	STRING	なし	有効な値の詳細については、「エンドポイント」をご参照ください。 VVR 8.0.8 以降でのみサポートされています。
  scan.incremental.close-idle-reader.enabled	スナップショット完了後にアイドルリーダーを閉じるかどうかを指定します。	いいえ	BOOLEAN	false	VVR 8.0.1 以降でのみサポートされています。 この設定を有効にするには、execution.checkpointing.checkpoints-after-tasks-finish.enabled を true に設定する必要があります。
  scan.read-changelog-as-append-only.enabled	チェンジログデータストリームを追加専用データストリームに変換するかどうかを指定します。	いいえ	BOOLEAN	false	有効な値： true：INSERT、DELETE、UPDATE_BEFORE、UPDATE_AFTER を含むすべてのタイプのメッセージが INSERT メッセージに変換されます。このオプションは、上流テーブルからの削除メッセージを保存する必要があるなど、特別なシナリオでのみ有効にしてください。 false (デフォルト)：すべてのタイプのメッセージがそのまま下流に送信されます。 説明 VVR 8.0.8 以降でのみサポートされています。
  scan.only.deserialize.captured.tables.changelog.enabled	増分フェーズで、指定されたテーブルの変更イベントのみを逆シリアル化するかどうかを指定します。	いいえ	BOOLEAN	VVR 8.x バージョンではデフォルト値は false です。 VVR 11.1 以降ではデフォルト値は true です。	有効な値： true：ターゲットテーブルの変更データのみを逆シリアル化して、バイナリログの読み取りを高速化します。 false (デフォルト)：すべてのテーブルの変更データを逆シリアル化します。 説明 VVR 8.0.7 以降でのみサポートされています。 VVR 8.0.8 以前を使用する場合、パラメーター名を debezium.scan.only.deserialize.captured.tables.changelog.enable に変更する必要があります。
  scan.parse.online.schema.changes.enabled	増分フェーズで、RDS ロックレス変更 DDL イベントの解析を試みるかどうかを指定します。	いいえ	BOOLEAN	false	有効な値： true：RDS ロックレス変更 DDL イベントを解析します。 false (デフォルト)：RDS ロックレス変更 DDL イベントを解析しません。 これは実験的な機能です。オンラインのロックレス変更を実行する前に、回復のために Flink ジョブのスナップショットを取得してください。 説明 VVR 11.1 以降でのみサポートされています。
  scan.incremental.snapshot.backfill.skip	スナップショット読み取りフェーズ中にバックフィルをスキップするかどうかを指定します。	いいえ	BOOLEAN	false	有効な値： true：スナップショット読み取りフェーズ中にバックフィルをスキップします。 false (デフォルト)：スナップショット読み取りフェーズ中にバックフィルをスキップしません。 バックフィルは単一チャンクのスナップショットクエリ中にのみ適用され、完全読み取りフェーズ全体をカバーするものではありません。バックフィルがスキップされると、各チャンクのスナップショットクエリはその時点での最新のテーブルデータを読み取ります。チャンクが読み取られた後に発生した更新は、完全読み取りフェーズ中にはマージされず、増分フェーズに入ってから Binlog から読み取られます。たとえば、chunk5 のスナップショット中に発生した chunk5 への更新は、chunk5 のスナップショットに直接反映されます。リーダーが chunk80 に進んだ後に chunk5 が更新された場合、その更新は後で増分フェーズ中に Binlog から適用されます。 重要 有効にすると、チャンクのスキャン中またはスキャン後に発生した変更は、増分フェーズで Binlog から配信され、重複する可能性があります。at-least-once セマンティクスのみが保証されます。下流の sink が主キーによるべき等書き込みをサポートしている場合にのみ有効にしてください。 説明 VVR 11.1 以降でのみサポートされています。
  scan.incremental.snapshot.unbounded-chunk-first.enabled	スナップショット読み取りフェーズ中に無制限チャンクを最初にディスパッチするかどうかを指定します。	いいえ	BOOELEAN	false	有効な値： true：スナップショット読み取りフェーズ中に無制限チャンクを最初にディスパッチします。 false (デフォルト)：スナップショット読み取りフェーズ中に無制限チャンクを最初にディスパッチしません。 これは実験的な機能です。有効にすると、スナップショットフェーズで最後のチャンクを同期する際の TaskManager での OOM エラーのリスクを軽減できます。ジョブの初回起動前にこのパラメーターを追加してください。 説明 VVR 11.1 以降でのみサポートされています。
  binlog.session.network.timeout	バイナリログ接続のネットワーク読み書きタイムアウト。	いいえ	DURATION	10m	0s に設定すると、MySQL サーバーのデフォルトのタイムアウトが使用されます。 説明 VVR 11.5 以降でのみサポートされています。
  scan.rate-limit.records-per-second	ソースが 1 秒あたりに送信するレコードの最大数を制限します。	いいえ	LONG	なし	これは、データ読み取りを制限する必要があるシナリオに適用されます。この制限は、完全フェーズと増分フェーズの両方で有効です。 ソースの numRecordsOutPerSecond メトリックは、データストリーム全体が 1 秒あたりに出力するレコード数を反映します。このメトリックに基づいてこのパラメーターを調整できます。 完全データ読み取りフェーズでは、通常、各バッチで読み取る行数を減らす必要があります。scan.incremental.snapshot.chunk.size パラメーターの値を減らすことができます。 説明 VVR 11.5 以降でのみサポートされています。
  scan.binlog.tolerate.gtid-holes	このパラメーターを有効にすると、GTID シーケンスのギャップが無視され、ジョブが不連続なイベントをバイパスして実行を継続できるようになります。	いいえ	BOOLEAN	false	このパラメーターを有効にする前に、ジョブの開始オフセットが期限切れになっていないことを確認する必要があります。ジョブがクリアされた、または期限切れの GTID オフセットから開始すると、エンジンは欠落しているログをサイレントにスキップし、データ損失につながります。 説明 このパラメーターは VVR 11.6 以降でのみサポートされています。
  scan.filter-unchanged-row.enabled	SQL で使用されていないフィールドの変更イベントをフィルタリングするかどうかを指定します。	いいえ	BOOLEAN	false	このパラメーターを true に設定すると、CDC ソーステーブルは UPDATE イベントを受信したときに、SQL ステートメントで参照されている列のみを比較します。参照されている列の値が変更されていない場合、イベントは下流に送信されません。このパラメーターは、複数ストリームの JOIN シナリオで有効です。上流テーブルが SQL ステートメントで参照されていないフィールドを更新した場合、不要な DELETE および INSERT イベントは生成されません。これにより、下流の演算子でのデータ増幅と結果のジッターが減少します。 説明 このパラメーターは VVR 11.8 以降でのみサポートされています。
  scan.projection-pushdown.enabled	プロジェクションプッシュダウンを有効にするかどうかを指定します。	いいえ	BOOLEAN	false	このパラメーターを true に設定すると、CDC ソーステーブルは SQL ステートメントで参照されている列のみを解析して送信します。参照されていない列はシリアル化されず、ネットワーク経由で送信されません。これにより、I/O と計算のオーバーヘッドが削減されます。このパラメーターは scan.filter-unchanged-row.enabled と一緒に使用すると、より効果的です。プロジェクションプッシュダウンがまずデータ量を削減し、その後、変更されていない行がフィルタリングされて、下流の演算子の処理負荷がさらに軽減されます。 説明 このパラメーターは VVR 11.8 以降でのみサポートされています。

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

  パラメーター	説明	必須	データの型	デフォルト値	注釈
  url	MySQL JDBC URL。	いいえ	STRING	なし	URL のフォーマットは次のとおりです：jdbc:mysql://<endpoint>:<port>/<database_name>。
  lookup.max-retries	データ読み取りに失敗した後の最大再試行回数。	いいえ	INTEGER	3	VVR 6.0.7 以降でのみサポートされています。
  lookup.cache.strategy	キャッシュポリシー。	いいえ	STRING	None	サポートされているキャッシュポリシーは None、LRU、ALL です。値の詳細については、「ディメンションテーブルの JOIN ステートメント」をご参照ください。 説明 LRU キャッシュポリシーを使用する場合、lookup.cache.max-rows パラメーターも設定する必要があります。
  lookup.cache.max-rows	キャッシュされる行の最大数。	いいえ	INTEGER	100000	LRU キャッシュポリシーを選択した場合、キャッシュサイズを設定する必要があります。 ALL キャッシュポリシーを選択した場合、キャッシュサイズを設定する必要はありません。
  lookup.cache.ttl	キャッシュの生存時間 (TTL)。	いいえ	DURATION	10 s	lookup.cache.ttl の設定は lookup.cache.strategy に依存します： lookup.cache.strategy が None に設定されている場合、lookup.cache.ttl を設定する必要はありません。これはキャッシュがタイムアウトしないことを意味します。 lookup.cache.strategy が LRU に設定されている場合、lookup.cache.ttl はキャッシュの TTL です。デフォルトでは、キャッシュは期限切れになりません。 lookup.cache.strategy が ALL に設定されている場合、lookup.cache.ttl はキャッシュのロード時間です。デフォルトでは、キャッシュは再読み込みされません。 1min や 10s のような時間形式を使用してください。
  lookup.max-join-rows	プライマリテーブルのレコードがディメンションテーブルのレコードと一致した場合に返される結果の最大数。	いいえ	INTEGER	1024	なし。
  lookup.filter-push-down.enabled	ディメンションテーブルのフィルタープッシュダウンを有効にするかどうかを指定します。	いいえ	BOOLEAN	false	有効な値： true：ディメンションテーブルのフィルタープッシュダウンを有効にします。MySQL データベーステーブルからデータをロードする際、ディメンションテーブルは SQL ジョブで設定された条件に基づいて事前にデータをフィルタリングします。 false (デフォルト)：ディメンションテーブルのフィルタープッシュダウンを無効にします。MySQL データベーステーブルからデータをロードする際、ディメンションテーブルはすべてのデータをロードします。 説明 VVR 8.0.7 以降でのみサポートされています。 重要 ディメンションテーブルのプッシュダウンは、Flink テーブルがディメンションテーブルとして使用される場合にのみ有効にすべきです。MySQL ソーステーブルはフィルタープッシュダウンの有効化をサポートしていません。Flink テーブルがソーステーブルとディメンションテーブルの両方として使用され、ディメンションテーブルでフィルタープッシュダウンが有効になっている場合、SQL ヒントを使用して、ソーステーブルに対してこの設定項目を明示的に false に設定する必要があります。そうしないと、ジョブが異常に実行される可能性があります。
  lookup.cache.cache-empty	空のクエリ結果をキャッシュするかどうかを指定します。	いいえ	BOOLEAN	true	キャッシュポリシーが LRU の場合にのみ有効です。有効な値： true (デフォルト)：クエリがデータを返さない場合、空の結果がキャッシュされ、後続のクエリに対して直接返されます。 false：クエリがデータを返さない場合、結果はキャッシュされず、次のクエリで再度データが検索されます。 説明 このパラメーターは VVR 11.9.0 以降でのみサポートされています。

• 結果テーブルのみ

  パラメーター	説明	必須	データの型	デフォルト値	注釈
  url	MySQL JDBC URL。	いいえ	STRING	なし	URL のフォーマットは次のとおりです：jdbc:mysql://<endpoint>:<port>/<database_name>。
  sink.max-retries	データ書き込みに失敗した後の最大再試行回数。	いいえ	INTEGER	3	なし。
  sink.buffer-flush.batch-size	単一のバッチ書き込みの行数。	いいえ	INTEGER	4096	なし。
  sink.buffer-flush.max-rows	メモリにキャッシュされるデータ行の数。	いいえ	INTEGER	10000	このパラメーターは、主キーが指定された後にのみ有効になります。
  sink.buffer-flush.interval	キャッシュをフラッシュする間隔。指定された待機時間の後、キャッシュ内のデータが出力条件を満たさない場合、システムはキャッシュ内のすべてのデータを自動的に出力します。	いいえ	DURATION	1s	なし。
  sink.ignore-delete	データの DELETE 操作を無視するかどうかを指定します。	いいえ	BOOLEAN	false	Flink SQL によって生成されたストリームに delete または update-before レコードが含まれている場合、複数の出力タスクが同じテーブルの異なるフィールドを同時に更新すると、データの不整合が発生する可能性があります。 たとえば、レコードが削除された後、別のタスクが一部のフィールドのみを更新します。更新されなかったフィールドは null またはデフォルト値になり、データエラーを引き起こします。 sink.ignore-delete を true に設定することで、上流の DELETE および UPDATE_BEFORE 操作を無視して、このような問題を回避できます。 説明 UPDATE_BEFORE は Flink のリトラクションメカニズムの一部であり、更新操作で古い値を「取り消す」ために使用されます。 ignoreDelete = true の場合、すべての DELETE および UPDATE_BEFORE タイプのレコードはスキップされます。INSERT および UPDATE_AFTER レコードのみが処理されます。
  sink.ignore-delete-mode	DELETE 操作が無視された後の delete タイプのレコードの処理戦略。	いいえ	STRING	ALL	有効な値： ALL：-D と -U の両方のレコードを無視します。 REAL_DELETE：-D レコードのみを無視します。 UPDATE_BEFORE：-U レコードのみを無視します。 説明 このオプションは Realtime Compute エンジン VVR 11.8 以降でのみサポートされています。 sink.ignore-delete=true の場合にのみ有効です。単独で設定するとエラーになります。
  sink.ignore-null-when-update	データを更新する際、入力データフィールドの値が null の場合に、対応するフィールドを null に更新するか、そのフィールドの更新をスキップするかを指定します。	いいえ	BOOLEAN	false	有効な値： true：フィールドを更新しません。このパラメーターは、Flink テーブルに主キーが設定されている場合にのみ true に設定できます。true に設定した場合： VVR 8.0.6 以前の場合、結果テーブルはバッチ書き込みをサポートしません。 VVR 8.0.7 以降の場合、結果テーブルはバッチ書き込みをサポートします。 バッチ書き込みは書き込み効率と全体のスループットを大幅に向上させることができますが、データ遅延と OOM エラーのリスクを伴います。したがって、ビジネスシナリオに基づいてトレードオフを行う必要があります。 false：フィールドを null に更新します。 説明 このパラメーターは VVR 8.0.5 以降でのみサポートされています。
  sink.force-batch-on-non-primary-table	主キーのない MySQL 結果テーブルにデータが書き込まれるときに、レコードを強制的にバッファリングし、JDBC バッチを使用してバッチで挿入するかどうかを指定します。	いいえ	BOOLEAN	false	このパラメーターは、主キーのないテーブルでのみ有効です。このパラメーターを true に設定すると、DELETE および UPDATE_BEFORE イベントはサイレントに破棄され、INSERT および UPDATE_AFTER イベントは INSERT ステートメントとしてバッチで書き込まれます。データが追加専用であるか、前述のセマンティクスが許容できる場合にのみ、このパラメーターを有効にすることを推奨します。 説明 このパラメーターは VVR 11.8 以降でのみサポートされています。 データが追加専用であるか、前述のセマンティクスが許容できる場合にのみ、このパラメーターを有効にしてください。そうしないと、データが不整合になる可能性があります。

型マッピング

• CDC ソーステーブル

  MySQL CDC フィールド型	Flink フィールド型
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  TINYINT UNSIGNED ZEROFILL
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  SMALLINT UNSIGNED ZEROFILL
  BIGINT	BIGINT
  INT UNSIGNED
  INT UNSIGNED ZEROFILL
  MEDIUMINT UNSIGNED
  MEDIUMINT UNSIGNED ZEROFILL
  BIGINT UNSIGNED	DECIMAL(20, 0)
  BIGINT UNSIGNED ZEROFILL
  SERIAL
  FLOAT [UNSIGNED] [ZEROFILL]	FLOAT
  DOUBLE [UNSIGNED] [ZEROFILL]	DOUBLE
  DOUBLE PRECISION [UNSIGNED] [ZEROFILL]
  REAL [UNSIGNED] [ZEROFILL]
  NUMERIC(p, s) [UNSIGNED] [ZEROFILL]	DECIMAL(p, s)
  DECIMAL(p, s) [UNSIGNED] [ZEROFILL]
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]	TIMESTAMP [(p)]
  	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
  CHAR(n)	STRING
  VARCHAR(n)
  TEXT
  BINARY	BYTES
  VARBINARY
  BLOB

  重要

  MySQL で TINYINT(1) 型を使用して 0 と 1 以外の値を保存しないでください。property-version=0 の場合、MySQL CDC ソーステーブルはデフォルトで TINYINT(1) を Flink の BOOLEAN 型にマッピングします。これにより、データの不正確さが生じる可能性があります。TINYINT(1) 型を使用して 0 と 1 以外の値を保存するには、設定パラメーター catalog.table.treat-tinyint1-as-boolean をご参照ください。

• ディメンションテーブルと結果テーブル

  MySQL フィールド型	Flink フィールド型
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  BIGINT	BIGINT
  INT UNSIGNED
  BIGINT UNSIGNED	DECIMAL(20, 0)
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DOUBLE PRECISION
  NUMERIC(p, s)	DECIMAL(p, s) 説明 ここで p <= 38。
  DECIMAL(p, s)
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]
  CHAR(n)	CHAR(n)
  VARCHAR(n)	VARCHAR(n)
  BIT(n)	BINARY(⌈n/8⌉)
  BINARY(n)	BINARY(n)
  VARBINARY(N)	VARBINARY(N)
  TINYTEXT	STRING
  TEXT
  MEDIUMTEXT
  LONGTEXT
  TINYBLOB	BYTES 重要 Flink は、2,147,483,647 (2^31 - 1) バイト以下の MySQL BLOB 型レコードのみをサポートします。
  BLOB
  MEDIUMBLOB
  LONGBLOB

データ遅延の診断とジョブスループットの最適化

増分フェーズでデータ遅延が発生した場合は、次の手順で問題を分析してください：

1. 概要ページの currentFetchEventTimeLag および currentEmitEventTimeLag メトリックを確認します。currentFetchEventTimeLag メトリックは、バイナリログからのデータ読み取りの遅延を表します。currentEmitEventTimeLag メトリックは、ジョブに関連するテーブルのバイナリログからのデータ読み取りの遅延を表します。

   シナリオ	説明
   currentFetchEventTimeLag は低いが、currentEmitEventTimeLag は高く、ほとんど更新されない。	低い currentFetchEventTimeLag は、データベースからのバイナリログのプルが効率的であることを示します。しかし、バイナリログにはジョブが読み取る必要のあるテーブルのデータがほとんど含まれていません。そのため、currentEmitEventTimeLag はほとんど更新されません。これは予期された動作です。
   currentFetchEventTimeLag と currentEmitEventTimeLag の両方が高い。	これは、ソーステーブルの読み取りパフォーマンスが低いことを示します。このセクションの以降の手順で最適化を進めることができます。

2. バックプレッシャーは、ソースが下流の演算子にデータを送信するレートを低下させる可能性があります。sourceIdleTime が定期的に増加し、currentFetchEventTimeLag と currentEmitEventTimeLag の両方が継続的に増加することが観察される場合があります。これを解決するには、バックプレッシャーが発生しているノードの並列度を上げてください。

3. CPU ページの TM CPU Usage メトリックと JVM ページの TM GC Time メトリックを確認して、CPU またはメモリリソースが不足していないか判断します。ジョブリソースを増やすことで読み取りパフォーマンスを最適化できます。また、ミニバッチパラメーターを有効にしてスループットを向上させることもできます。詳細については、「高性能 Flink SQL 最適化技術」をご参照ください。

4. ジョブに大きな状態を持つ SinkUpsertMaterializer 演算子が存在する場合、読み取りパフォーマンスに影響を与える可能性があります。ジョブの並列度を上げるか、SinkUpsertMaterializer 演算子を避けることを検討してください。詳細については、「SinkUpsertMaterializer の使用を避ける」をご参照ください。既存のジョブから SinkUpsertMaterializer 演算子を削除するには、ステートレス再起動が必要です。これは、ジョブトポロジーが変更されるため、既存の状態から開始するとジョブの失敗やデータ損失を引き起こす可能性があるためです。

チャンクパラメーターの最適化

チャンクデータとメタデータはメモリに保存されるため、メモリ不足 (OOM) エラーが発生することがあります。OOM エラーが発生したコンポーネントに応じてパラメーターを調整できます：

• JobManager

  JobManager はすべてのチャンクのメタデータを保存します。チャンク数が多すぎると OOM エラーが発生する可能性があります。これを解決するには、scan.incremental.snapshot.chunk.size の値を増やしてチャンク数を減らします。または、ランタイム設定で jobmanager.memory.heap.size を設定して JobManager のヒープメモリを増やすこともできます。詳細については、「Flink パラメーター設定」をご参照ください。

• TaskManager

  • TaskManager は各チャンクのデータを読み取ります。チャンクに含まれる行が多すぎると、OOM エラーが発生する可能性があります。これを解決するには、scan.incremental.snapshot.chunk.size の値を減らしてチャンクあたりの行数を減らします。また、ランタイム設定で TaskManager Memory の値を増やして TaskManager のヒープメモリを増やすこともできます。

  • VVR 8.0.8 以前では、最後のチャンクに大量のデータが含まれる可能性があり、TaskManager で OOM エラーが発生することがあります。この問題を回避するために、VVR 8.0.9 以降へのアップグレードを推奨します。

  • 複合主キーを持つ MySQL CDC ソーステーブルの場合、チャンクはデフォルトでキーの最初の列に基づいて分割されます。データが大幅に偏っており、多くの行がその列で同じ値を共有している場合、その値のチャンクが非常に大きくなり、TaskManager で OOM エラーが発生する可能性があります。scan.incremental.snapshot.chunk.key-column を設定して、主キーから別の列を分割用に指定できます。