Realtime Compute for Apache Flink:MySQL

SQL

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

構文

CREATE TEMPORARY TABLE mysql_sink (
   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 を指定できます。これら2つの値は同等です。ディメンションテーブルまたはシンクテーブルの場合、値は mysql である必要があります。
  hostname	MySQL データベースの IP アドレスまたはホスト名。	はい	STRING	-	VPC アドレスを推奨します。 説明 MySQL データベースと Realtime Compute for Apache Flink ワークスペースが異なる VPC にある場合、VPC 間のネットワーク接続を確立するか、パブリックネットワーク経由で接続する必要があります。詳細については、「Space Management and Operations」および「How can a fully managed Flink cluster access the public network?」をご参照ください。
  username	MySQL データベースのユーザー名。	はい	STRING	-	-
  password	MySQL データベースのパスワード。	はい	STRING	-	-
  database-name	データベース名。	はい	STRING	-	ソースとして使用する場合、正規表現を指定して複数のデータベースからデータを読み取ることができます。 正規表現を使用する場合、先頭と末尾の一致には ^ および $ 記号を使用しないでください。詳細については、table-name の備考をご参照ください。
  table-name	テーブル名。	はい	STRING	-	ソースとして使用する場合、正規表現を指定して複数のテーブルからデータを読み取ることができます。 複数の MySQL テーブルから読み取る際のパフォーマンスと効率を向上させるには、複数の CTAS ステートメントを単一のジョブとして送信してください。これにより、複数のバイナリログリスナーの有効化を防ぐことができます。詳細については、「Run multiple CTAS statements in a single job」をご参照ください。 正規表現を使用する場合、先頭と末尾の一致には ^ および $ 記号を使用しないでください。具体的な理由については、以下の説明をご参照ください。 説明 MySQL CDC ソーステーブルが正規表現を使用してテーブル名を照合する場合、入力された database-name と table-name を文字列 \\. (VVR バージョン 8.0.1 より前の場合は文字 . を使用します) と連結してフルパス正規表現を作成し、この式を使用して MySQL データベース内のテーブルの完全修飾名と照合します。 たとえば、'database-name'='db_.*' および 'table-name'='tb_.+' を構成すると、コネクタは正規表現 db_.*\\.tb_.+ (8.0.1 より前のバージョンでは db_.*.tb_.+) を使用して完全修飾テーブル名と照合し、読み取るテーブルを決定します。
  port	MySQL データベースのポート。	いいえ	INTEGER	3306	-

• ソース固有

  パラメーター	説明	必須	タイプ	デフォルト	備考
  server-id	データベースクライアントの数値 ID です。	いいえ	STRING	5400～6400 の範囲でランダムに生成される整数。	サーバー ID は、MySQL クラスター内でグローバルに一意である必要があります。同じデータベースに接続するジョブごとに異なる ID を設定してください。 このパラメーターは、5400-6400 のような ID 範囲もサポートしています。同時リーダーを使用してインクリメンタルスナップショットを有効にする場合は、各リーダーが一意のサーバー ID を持つように ID 範囲を指定してください。範囲内の ID 数は、ソースの並列度以上である必要があります。詳細については、「Server ID Usage」をご参照ください。
  scan.incremental.snapshot.enabled	インクリメンタルスナップショットを有効にするかどうかを指定します。	いいえ	BOOLEAN	true	デフォルトでは、インクリメンタルスナップショットが有効になっています。インクリメンタルスナップショットは、テーブル全体のスナップショットを読み取る仕組みです。従来のスナップショット方式と比較して、次の利点があります： ソースはスナップショットを並列に読み取れます。 ソースは、スナップショットフェーズ中にチャンクレベルのチェックポイントをサポートします。 ソースでは、グローバル読み取りロック (FLUSH TABLES WITH READ LOCK) は不要です。 並列読み取りを有効にするには、同時リーダーごとに一意のサーバー ID が必要です。そのため、server-id は 5400-6400 のような範囲に設定し、範囲のサイズはソースの並列度以上である必要があります。 説明 このパラメーターは Flink エンジン 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	開始オフセットのバイナリログファイル名です。	いいえ	STRING	-	この設定を使用する場合、scan.startup.mode を specific-offset に設定する必要があります。ファイル名の形式は mysql-bin.000003 などです。
  scan.startup.specific-offset.pos	指定したバイナリログファイル内の開始オフセットです。	いいえ	INTEGER	-	この設定を使用する場合、scan.startup.mode を specific-offset に設定する必要があります。
  scan.startup.specific-offset.gtid-set	開始オフセットの GTID セットです。	いいえ	STRING	-	この設定を使用する場合、scan.startup.mode を specific-offset に設定する必要があります。GTID セットの形式は 24DA167-0C0C-11E8-8442-00059A3C7B00:1-19 などです。
  scan.startup.timestamp-millis	開始タイムスタンプ (ミリ秒) です。	いいえ	LONG	-	この設定を使用する場合、scan.startup.mode を timestamp に設定する必要があります。タイムスタンプの単位はミリ秒です。 重要 指定したタイムスタンプから開始する場合、MySQL CDC コネクターは各バイナリログファイルの初期イベントをスキャンして、タイムスタンプに対応するファイルを特定します。指定したタイムスタンプに対応するバイナリログファイルがパージされておらず、データベースサーバー上でアクセス可能であることを確認してください。
  server-time-zone	データベースサーバーのセッションタイムゾーンです。	いいえ	STRING	このパラメーターを指定しない場合、システムは Flink ジョブの実行環境のタイムゾーン (選択したアベイラビリティーゾーンのタイムゾーン) を使用します。	例：Asia/Shanghai。このパラメーターは、TIMESTAMP の値を文字列に変換する方法を制御します。詳細については、「Debezium temporal values」をご参照ください。
  debezium.min.row.count.to.stream.results	テーブルの行数がこの値を超えると、コネクターはスナップショット読み取りではなくバッチ読み取りを使用します。	いいえ	INTEGER	1000	Flink は、MySQL ソーステーブルを次のいずれかの方法で読み取ります： スナップショット読み取り：テーブル全体をメモリに読み込みます。この方法は高速ですがメモリ消費が大きく、大きなテーブルではメモリ不足 (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	ハートビートイベントはバイナリログのオフセットを進めます。これは、更新頻度が低いテーブルなどでバイナリログのオフセットが自動的に進まない場合に特に有効です。これにより、バイナリログのオフセットが古くなることを防ぎ、ジョブが失敗してステートレス再起動が必要になる事態を回避できます。
  scan.incremental.snapshot.chunk.key-column	スナップショットフェーズ中にチャンクを分割するために使用する列を指定します。	備考欄をご参照ください。	STRING	-	プライマリーキーがないテーブルでは必須です。選択する列は非 NULL タイプ (NOT NULL) である必要があります。 プライマリーキーがあるテーブルでは任意です。プライマリーキーから 1 列のみ選択できます。
  rds.region-id	ApsaraDB RDS for MySQL インスタンスのリージョン ID です。	OSS からアーカイブログを読み取る場合は必須です。	STRING	-	リージョン ID の一覧については、「Regions and zones」をご参照ください。 重要 MySQL CDC はランダムな GTID 文字列を生成します。これはバイナリログファイルのオフセットのように単調増加しません。そのため、GTID に基づいて特定のファイルを特定するには、コネクターが OSS からすべてのアーカイブログをダウンロードして解析する必要があり、リソース消費が大きく時間もかかります。したがって、GTID ベースの起動モードではアーカイブされた OSS ログからの読み取りは実行できません。この機能は、特定のタイムスタンプまたはバイナリログファイルのオフセットからの起動のみをサポートします。また、GTID に依存するため、アーカイブログ内で主従切り替えが発生するシナリオもサポートしていません。この機能を使用する前に、これらの制限を十分に評価してください。
  rds.access-key-id	ApsaraDB RDS for MySQL アカウントの AccessKey ID です。	OSS からアーカイブログを読み取る場合は必須です。	STRING	-	詳細については、「How do I view my AccessKey ID and AccessKey secret?」をご参照ください。 重要 セキュリティに関する注意：認証情報の露出を避けるため、変数管理を使用して AccessKey ID を保存し、取得してください。
  rds.access-key-secret	ApsaraDB RDS for MySQL アカウントの AccessKey シークレットです。	OSS からアーカイブログを読み取る場合は必須です。	STRING	-	詳細については、「How do I view my AccessKey ID and AccessKey secret?」をご参照ください。 重要 セキュリティに関する注意：認証情報の露出を避けるため、変数管理を使用して AccessKey シークレットを保存し、取得してください。
  rds.db-instance-id	ApsaraDB RDS for MySQL インスタンスのインスタンス ID です。	OSS からアーカイブログを読み取る場合は必須です。	STRING	-	-
  rds.main-db-id	ApsaraDB RDS for MySQL インスタンスのプライマリデータベース ID です。	いいえ	STRING	-	プライマリデータベース ID の取得方法については、「ApsaraDB RDS for MySQL log backup」をご参照ください。 Flink エンジン VVR 8.0.7 以降でサポートされています。 説明 このパラメーターを指定しない場合、Flink エンジン VVR 11.7 以降では ApsaraDB RDS for MySQL の接続情報に基づいてプライマリデータベース ID が自動的にクエリされます。
  rds.download.timeout	OSS から単一のアーカイブログファイルをダウンロードする際のタイムアウトです。	いいえ	DURATION	60s	-
  rds.endpoint	OSS からバイナリログ情報を取得するためのサービスエンドポイントです。	いいえ	STRING	-	使用可能な値の一覧については、「Service endpoints」をご参照ください。 Flink エンジン VVR 8.0.8 以降でサポートされています。
  scan.incremental.close-idle-reader.enabled	スナップショットフェーズの完了後にアイドルリーダーをクローズするかどうかを指定します。	いいえ	BOOLEAN	false	Flink エンジン 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 (デフォルト)：すべてのメッセージタイプをそのまま出力します。 説明 Flink エンジン VVR 8.0.8 以降でサポートされています。
  scan.only.deserialize.captured.tables.changelog.enabled	増分フェーズ中に、キャプチャ対象テーブルの変更イベントのみをデシリアライズするかどうかを指定します。	いいえ	BOOLEAN	Flink エンジン VVR 8.x では false 。 Flink エンジン VVR 11.1 以降では true 。	有効な値： true：キャプチャ対象テーブルの変更データのみをデシリアライズするため、バイナリログの読み取り速度が向上します。 false：すべてのテーブルの変更データをデシリアライズします。 説明 Flink エンジン VVR 8.0.7 以降でサポートされています。 Flink エンジン VVR 8.0.8 以前では、このパラメーター名は debezium.scan.only.deserialize.captured.tables.changelog.enable です。
  scan.parse.online.schema.changes.enabled	増分フェーズ中に、RDS lockless change の DDL イベントを解析するかどうかを指定します。	いいえ	BOOLEAN	false	有効な値： true：RDS lockless change の DDL イベントを解析します。 false (デフォルト)：RDS lockless change の DDL イベントを解析しません。 これは実験的な機能です。オンラインのロックレス変更を実行する前に Flink ジョブのセーブポイントを取得し、必要に応じて回復できるようにしてください。 説明 Flink エンジン VVR 11.1 以降でサポートされています。
  scan.incremental.snapshot.backfill.skip	スナップショットフェーズ中にバックフィルプロセスをスキップするかどうかを指定します。	いいえ	BOOLEAN	false	有効な値： true：バックフィルプロセスをスキップします。 false (デフォルト)：バックフィルプロセスをスキップしません。 バックフィルをスキップした場合、スナップショットフェーズ中に発生したテーブルの変更は、スナップショットにマージされず、増分フェーズで後から読み取られます。 重要 バックフィルをスキップすると、スナップショットフェーズ中に発生した変更がリプレイされる可能性があるため、データの不整合につながる可能性があります。この場合、at-least-once セマンティクスのみが提供されます。 説明 Flink エンジン VVR 11.1 以降でサポートされています。
  scan.incremental.snapshot.unbounded-chunk-first.enabled	スナップショットフェーズ中に境界なしチャンクを優先して分配するかどうかを指定します。	いいえ	BOOLEAN	false	有効な値： true：スナップショットフェーズ中に境界なしチャンクの分配を優先します。 false (デフォルト)：境界なしチャンクの分配を優先しません。 これは実験的な機能です。有効にすると、TaskManager が最後のチャンクを同期する際のメモリ不足 (OOM) エラーのリスクを低減できます。この機能は、ジョブの初回起動前に有効にしてください。 説明 Flink エンジン VVR 11.1 以降でサポートされています。
  binlog.session.network.timeout	バイナリログ接続のネットワーク読み取り/書き込みタイムアウトです。	いいえ	DURATION	10m	0s に設定すると、接続は MySQL サーバーのデフォルトタイムアウトを使用します。 説明 Flink エンジン VVR 11.5 以降でサポートされています。
  scan.rate-limit.records-per-second	ソースが 1 秒あたりに出力できるレコード数の上限を設定します。	いいえ	LONG	-	このパラメーターを使用してデータの読み取りレートを制限します。この制限は、スナップショットフェーズと増分フェーズの両方に適用されます。 ソースの numRecordsOutPerSecond メトリクスは、データストリーム全体で 1 秒あたりに出力されたレコード数を示します。このメトリクスに基づいて、このパラメーターを調整できます。 スナップショット読み取りフェーズでは、通常、scan.incremental.snapshot.chunk.size パラメーターの値を小さくして、各バッチで読み取るデータ行数を減らす必要があります。 説明 Flink エンジン VVR 11.5 以降でサポートされています。
  scan.binlog.tolerate.gtid-holes	GTID シーケンスのギャップを無視し、不連続なイベントをスキップしてジョブを継続できるようにします。	いいえ	BOOLEAN	false	このパラメーターを有効にする前に、ジョブの起動オフセットが期限切れになっていないことを確認してください。ジョブがパージ済みまたは期限切れの GTID オフセットから開始すると、エンジンは欠落したログを暗黙的にスキップし、データ損失が発生します。 説明 Flink エンジン VVR 11.6 以降でサポートされています。

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

  パラメーター	説明	必須	タイプ	デフォルト	備考
  url	MySQL データベースの JDBC URL。	いいえ	文字列	なし	URL の形式は次のとおりです： jdbc:mysql://<endpoint>:<port>/<database name>。
  lookup.max-retries	ルックアップリクエストが失敗した場合の最大リトライ回数。	いいえ	整数	3	Flink コンピューティングエンジン VVR 6.0.7 以降でのみサポートされます。
  lookup.cache.strategy	キャッシュポリシー。	いいえ	文字列	None	サポートされる値： None、LRU、ALL。詳細については、「Background information」をご参照ください。 説明 LRU キャッシュポリシーには、lookup.cache.max-rows パラメーターが必要です。
  lookup.cache.max-rows	キャッシュに格納する最大行数。	いいえ	整数	100000	キャッシュポリシーが LRU に設定されている場合は必須です。 キャッシュポリシーが ALL に設定されている場合はオプションです。
  lookup.cache.ttl	キャッシュされたエントリの Time-to-Live (TTL)。	いいえ	期間	なし	lookup.cache.ttl パラメーターの動作は、lookup.cache.strategy の値によって異なります: lookup.cache.strategy を None に設定した場合、lookup.cache.ttl を設定する必要はありません。これは、キャッシュの有効期限が切れないことを意味します。 lookup.cache.strategy が LRU に設定されている場合、lookup.cache.ttl はキャッシュのタイムアウト期間を指定します。デフォルトでは、キャッシュは無期限です。 lookup.cache.strategy が ALL に設定されている場合、lookup.cache.ttl はキャッシュの再読み込み間隔を指定します。デフォルトでは、キャッシュは再読み込みされません。 期間は、1min や 10s などの時間形式で指定します。
  lookup.max-join-rows	ソーステーブルの各行に対してディメンションテーブルから返す一致する行の最大数。	いいえ	整数	1024	なし
  lookup.filter-push-down.enabled	ディメンションテーブルのフィルタープッシュダウンを有効にするかどうかを指定します。	いいえ	ブール値	false	有効な値： true: フィルタープッシュダウンを有効にします。システムは、データをロードする前に SQL ジョブの条件に基づいてディメンションテーブルのデータをフィルター処理します。 false (デフォルト)：フィルタープッシュダウンを無効にします。システムは、ディメンションテーブルからデータセット全体をロードします。 説明 Flink コンピューティングエンジン 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	各バッチ書き込みに含めるレコード数です。	いいえ	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 のリトラクションメカニズムの一部であり、更新操作中に古い値をリトラクトします。 sink.ignore-delete が true の場合、シンクはすべての DELETE および UPDATE_BEFORE レコードをスキップし、INSERT および UPDATE_AFTER レコードのみを処理します。
  sink.ignore-null-when-update	更新レコード内の null 値の処理方法です。	いいえ	BOOLEAN	false	有効な値： true：null 値を無視し、宛先テーブルの対応フィールドは変更しません。この設定では、Flink テーブルにプライマリキーを定義する必要があります。true に設定した場合： VVR 8.0.6 以前では、シンクテーブルでバッチ書き込みはサポートされていません。 VVR 8.0.7 以降では、シンクテーブルでバッチ書き込みがサポートされています。 バッチ書き込みはスループットを大幅に向上できますが、データレイテンシーが発生したり、メモリ不足エラーのリスクが高まったりする可能性があります。ビジネス要件に基づいて、これらのトレードオフを評価してください。 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 以外の値を格納することは避けてください。デフォルトでは、プロパティバージョンが 0 の場合、MySQL CDC ソーステーブルは TINYINT(1) を Flink の BOOLEAN 型にマッピングするため、データの不正確さにつながる可能性があります。TINYINT(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

データインジェスト

MySQL コネクタを Flink CDC のデータソースとして使用できます。

構文

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 データインジェストジョブでは、[Catalogs] ページから組み込みの 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 設定で明示的に定義します。YAML で定義されたパラメーターが優先されます。

型マッピング

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