Realtime Compute for Apache Flink:MySQL

SQL

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

構文

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	なし	Virtual Private Cloud (VPC) のアドレスを指定することを推奨します。 説明 MySQL データベースと Realtime Compute for Apache Flink が同じ Virtual Private Cloud (VPC) にない場合は、VPC 間ネットワーク接続を確立するか、パブリックエンドポイントを使用してデータベースにアクセスする必要があります。詳細については、「ワークスペースの管理と運用」および「フルマネージド Flink クラスターからインターネットにアクセスする方法」をご参照ください。
  username	MySQL データベースサービスのユーザー名。	はい	STRING	なし	なし。
  password	MySQL データベースサービスのパスワード。	はい	STRING	なし	なし。
  database-name	MySQL データベース名。	はい	STRING	なし	データベースをソーステーブルとして使用する場合、データベース名に正規表現を使用して複数のデータベースからデータを読み取ることができます。 正規表現を使用する場合、文字列の先頭と末尾に一致させるために ^ および $ 記号を使用しないでください。詳細については、table-name パラメータの備考をご参照ください。
  table-name	MySQL テーブル名。	はい	STRING	なし	ソーステーブル名に正規表現を使用して複数のテーブルからデータを読み取ることができます。 複数の MySQL テーブルからデータを読み取る場合は、複数の CTAS ステートメントを 1 つのジョブとして送信してください。これにより、複数のバイナリログリスナーを有効にする必要がなくなり、パフォーマンスと効率が向上します。詳細については、「複数の CTAS ステートメント：単一ジョブとして送信」をご参照ください。 正規表現を使用する場合、文字列の先頭と末尾に一致させるために ^ および $ 記号を使用しないでください。詳細については、次の注をご参照ください。 説明 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 範囲を設定してください。詳細については、「Use Server ID」をご参照ください。
  scan.incremental.snapshot.enabled	増分スナップショットを有効にするかどうかを指定します。	いいえ	BOOLEAN	true	増分スナップショットはデフォルトで有効です。増分スナップショットは、全データのスナップショットを読み取るための新しい仕組みです。従来のスナップショット読み取り方法と比べて、増分スナップショットには次のような多くの利点があります： ソースは全データを並列に読み取ることができます。 ソースは、全データの読み取り時にチャンクレベルのチェックポイントをサポートします。 ソースは、全データの読み取り時にグローバル読み取りロック (FLUSH TABLES WITH read lock) を取得する必要がありません。 ソースで同時読み取りをサポートする場合、同時リーダーごとに一意の server ID が必要です。そのため、server-id は 5400-6400 などの範囲である必要があり、範囲のサイズは並列度以上である必要があります。 説明 この設定項目は、Ververica Runtime (VVR) 11.1 以降で削除されています。
  scan.incremental.snapshot.chunk.size	各チャンクのサイズ (行数) を指定します。	いいえ	INTEGER	8096	増分スナップショット読み取りが有効な場合、テーブルは複数のチャンクに分割されて読み取られます。チャンクのデータは、完全に読み取られるまでメモリにキャッシュされます。 各チャンクに含まれる行数が少ないほど、テーブル内のチャンクの総数は多くなります。これにより障害復旧の粒度は細かくなりますが、メモリ不足 (OOM) エラーが発生する可能性があり、全体のスループットが低下する場合があります。そのため、トレードオフを考慮し、適切なチャンクサイズを設定する必要があります。
  scan.snapshot.fetch.size	テーブルの全データを読み取る際に、1 回にプルするレコードの最大数を指定します。	いいえ	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 temporal values」をご参照ください。
  debezium.min.row.count.to.stream.results	テーブルの行数がこの値を超える場合、バッチ読み取りモードが使用されます。	いいえ	INTEGER	1000	Flink は、MySQL ソーステーブルから次のいずれかの方法でデータを読み取ります： フル読み取り：テーブル全体のデータを直接メモリに読み取ります。この方法は高速ですが、それに応じたメモリを消費します。ソーステーブルが非常に大きい場合、OOM エラーが発生するリスクがあります。 バッチ読み取り：一定数の行を 1 バッチとして複数回に分けて読み取り、すべてのデータを読み取るまで繰り返します。この方法は、大きなテーブルの読み取り時に 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 Configuration Properties」をご参照ください。
  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	なし	プライマリキーのないテーブルでは必須です。選択するカラムは非 NULL 型 (NOT NULL) である必要があります。 プライマリキーのあるテーブルでは任意です。プライマリキーから選択できるカラムは 1 つのみです。
  rds.region-id	Alibaba Cloud ApsaraDB RDS for MySQL インスタンスのリージョン ID を指定します。	OSS からアーカイブログを読み取る場合に必須です。	STRING	なし	リージョン ID の詳細については、「Regions and zones」をご参照ください。 重要 MySQL CDC の GTID 文字列はランダムに生成され、バイナリログファイルのオフセットのように単調増加しないため、ファイル内の GTID を特定するには OSS からすべてのアーカイブログをダウンロードして解析する必要があります。この処理は大量のリソースを消費し、時間もかかるため、GTID オフセットに依存する機能は実現できません。そのため、OSS のアーカイブログ機能は、指定したタイムスタンプ、または指定したバイナリログファイルオフセットからの開始のみをサポートします。指定した GTID からの開始はサポートしていません。また、アーカイブログ内でのプライマリ/セカンダリ切り替えのシナリオもサポートしていません。MySQL のプライマリ/セカンダリ切り替えは GTID に依存しているためです。本機能は、使用前に十分に評価してください。
  rds.access-key-id	Alibaba Cloud ApsaraDB RDS for MySQL アカウントの AccessKey ID を指定します。	OSS からアーカイブログを読み取る場合に必須です。	STRING	なし	詳細については、「How do I view the AccessKey ID and AccessKey secret?」をご参照ください。 重要 AccessKey 情報の漏えいを防ぐため、シークレット管理機能を使用して AccessKey ID を指定してください。詳細については、「Manage variables」をご参照ください。
  rds.access-key-secret	Alibaba Cloud ApsaraDB RDS for MySQL アカウントの AccessKey Secret を指定します。	OSS からアーカイブログを読み取る場合に必須です。	STRING	なし	詳細については、「How do I view the AccessKey ID and AccessKey secret?」をご参照ください。 重要 AccessKey 情報の漏えいを防ぐため、シークレット管理機能を使用して AccessKey Secret を指定してください。詳細については、「Manage variables」をご参照ください。
  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 log backup」をご参照ください。 VVR 8.0.7 以降でのみサポートされています。 説明 このパラメーターを指定しない場合、VVR 11.7 以降では ApsaraDB RDS for MySQL の接続情報に基づいてプライマリデータベース番号が自動的に照会されます。
  rds.download.timeout	OSS から単一のアーカイブログをダウンロードする際のタイムアウト期間を指定します。	いいえ	DURATION	60s	なし。
  rds.endpoint	OSS のバイナリログ情報を取得するためのサービスエンドポイントを指定します。	いいえ	STRING	なし	有効な値の詳細については、「Endpoints」をご参照ください。 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 (デフォルト)：スナップショット読み取りフェーズ中のバックフィルをスキップしません。 バックフィルをスキップすると、スナップショットフェーズ中に発生したテーブルの変更は、スナップショットにマージされず、後続の増分フェーズで読み取られます。 重要 バックフィルをスキップすると、スナップショットフェーズ中に発生した変更が再生される可能性があるため、データの不整合が発生する可能性があります。at-least-once セマンティクスのみが保証されます。 説明 VVR 11.1 以降でのみサポートされています。
  scan.incremental.snapshot.unbounded-chunk-first.enabled	スナップショット読み取りフェーズ中に、無制限チャンクを先にディスパッチするかどうかを指定します。	いいえ	BOOLEAN	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 以降でのみサポートされています。

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

  パラメーター	説明	必須	データ型	デフォルト値	備考
  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	キャッシュの Time-to-Live (TTL) です。	いいえ	DURATION	なし	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 に設定する必要があります。 そうしないと、ジョブが異常に実行される可能性があります。

• シンクテーブルのみ

  パラメーター	説明	必須	データ型	デフォルト値	備考
  url	MySQL JDBC URL。	任意	STRING	なし	URL 形式： jdbc:mysql://<endpoint>:<port>/<database_name>。
  sink.max-retries	データ書き込み失敗時の最大リトライ回数。	任意	INTEGER	3	なし。
  sink.buffer-flush.batch-size	1 回のバッチ書き込みにおける行数。	任意	INTEGER	4096	なし。
  sink.buffer-flush.max-rows	メモリにキャッシュされるデータ行数。	任意	INTEGER	10000	このパラメーターは、プライマリキーを指定した場合にのみ有効です。
  sink.buffer-flush.interval	キャッシュをフラッシュする間隔。指定した待機時間が経過してもキャッシュ内のデータが出力条件を満たさない場合、システムはキャッシュ内のすべてのデータを自動的に出力します。	任意	DURATION	1 s	なし。
  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-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 以降でのみサポートされます。

型マッピング

• 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

データインジェスト

データインジェストの YAML ジョブでは、MySQL コネクタをデータソースとして使用できます。

構文

source:
   type: mysql
   name: MySQL Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: xxx

構成アイテム

既存カタログの再利用

VVR 11.5 以降では、Flink CDC データインジェスト ジョブで、[Data Management] ページで作成した組み込み MySQL カタログを直接参照できます。これにより、接続プロパティを手動で記述する手間を削減できます。

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog

現在、データインジェスト ジョブは、次の MySQL カタログ パラメーターを自動的に再利用できます：

• hostname

• port

• username

• password

• catalog.table.metadata-columns

• catalog.table.treat-tinyint1-as-boolean

これらの自動的に再利用されるパラメーターをオーバーライドする場合は、対応する YAML パラメーターを明示的に記述できます。明示的に記述したパラメーターの方が優先度が高くなります。

型マッピング

次の表に、データインジェストにおけるデータ型マッピングを示します。