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	なし	VPC アドレスを入力することを推奨します。 説明 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 テーブルからデータを読み取る場合、複数の CTAS 文を単一のジョブとして投入します。これにより、複数のバイナリログリスナーが有効になるのを防ぎ、パフォーマンスと効率が向上します。詳細については、「複数の CTAS 文: 単一のジョブとして投入する」をご参照ください。 正規表現を使用する場合、文字列の開始と終了を一致させるために ^ および $ 記号を使用しないでください。詳細については、以下の注記をご参照ください。 説明 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-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 (デフォルト)：最初の起動時に完全な既存データをスキャンし、その後最新の binlog データを読み取ります。 latest-offset：最初の起動時に既存データをスキャンしません。binlog の末尾から読み取りを開始します。つまり、コネクタの起動後に行われた最新の変更のみを読み取ります。 earliest-offset：既存データをスキャンしません。利用可能な最も古い binlog から読み取りを開始します。 specific-offset：既存データをスキャンしません。指定した特定の binlog オフセットから開始します。scan.startup.specific-offset.file と scan.startup.specific-offset.pos の両方を設定するか、scan.startup.specific-offset.gtid-set のみを設定して特定の GTID セットから開始することでオフセットを指定できます。 timestamp：既存データをスキャンしません。指定したタイムスタンプから binlog の読み取りを開始します。タイムスタンプは scan.startup.timestamp-millis でミリ秒単位で指定します。 重要 earliest-offset、specific-offset、または timestamp 起動モードを使用する場合、スキーマの不一致によるエラーを避けるために、指定された binlog 消費位置とジョブの起動時刻の間で対応するテーブルのスキーマが変更されていないことを確認してください。
  scan.startup.specific-offset.file	特定オフセット起動モードを使用する場合の開始オフセットの binlog ファイル名。	いいえ	STRING	なし	この設定を使用する場合、scan.startup.mode を specific-offset に設定する必要があります。ファイル名の形式は、たとえば mysql-bin.000003 です。
  scan.startup.specific-offset.pos	特定オフセット起動モードを使用する場合に、指定された binlog ファイル内で開始するオフセット。	いいえ	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 は各 binlog ファイルの初期イベントを読み取ってそのタイムスタンプを判断し、指定された時間に対応する binlog ファイルを特定しようとします。指定されたタイムスタンプの binlog ファイルがデータベースからクリアされておらず、読み取り可能であることを確認してください。
  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 が binlog を読み取るためのカスタムオプション。	いいえ	STRING	なし	カスタム Debezium オプションを渡すことができます。たとえば、'debezium.event.deserialization.failure.handling.mode'='ignore' を使用して、解析エラーの処理ロジックを指定します。
  heartbeat.interval	ソースがハートビートイベントを使用して binlog オフセットを進める間隔。	いいえ	DURATION	30s	ハートビートイベントは、ソースの binlog オフセットを進めるために使用され、MySQL の更新が遅いテーブルに非常に役立ちます。このようなテーブルでは、binlog オフセットは自動的に進みません。ハートビートイベントは binlog オフセットを前進させ、期限切れの binlog オフセットが原因でジョブが失敗し、ステートレスな再起動が必要になる問題を回避します。
  scan.incremental.snapshot.chunk.key-column	スナップショットフェーズ中にチャンクを分割するために使用される列。	備考を参照	STRING	なし	プライマリキーのないテーブルでは必須です。選択された列は非ヌル型 (NOT NULL) でなければなりません。 プライマリキーのあるテーブルではオプションです。プライマリキーから 1 つの列のみを選択できます。
  rds.region-id	ApsaraDB RDS for MySQL インスタンスのリージョン ID。	OSS からアーカイブされたログを読み取る場合に必須です。	STRING	なし	リージョン ID の詳細については、「リージョンとゾーン」をご参照ください。
  rds.access-key-id	ApsaraDB RDS for MySQL インスタンスのアカウントの AccessKey ID。	OSS からアーカイブされたログを読み取る場合に必須です。	STRING	なし	詳細については、「AccessKey ID と AccessKey Secret の表示方法」をご参照ください。 重要 AccessKey 情報の漏洩を防ぐため、シークレット管理を使用して AccessKey ID を管理することを推奨します。詳細については、「変数の管理」をご参照ください。
  rds.access-key-secret	ApsaraDB RDS for MySQL インスタンスのアカウントの AccessKey Secret。	OSS からアーカイブされたログを読み取る場合に必須です。	STRING	なし	詳細については、「AccessKey ID と AccessKey Secret の表示方法」をご参照ください。 重要 AccessKey 情報の漏洩を防ぐため、シークレット管理を使用して AccessKey Secret を管理することを推奨します。詳細については、「変数の管理」をご参照ください。
  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 のログバックアップ」をご参照ください。 VVR 8.0.7 以降でのみサポートされます。
  rds.download.timeout	OSS から単一のアーカイブ済みログをダウンロードするためのタイムアウト期間。	いいえ	DURATION	60s	なし。
  rds.endpoint	OSS binlog 情報を取得するためのサービスエンドポイント。	いいえ	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：ターゲットテーブルの変更データのみを逆シリアル化して、binlog の読み取りを高速化します。 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 (デフォルト)：スナップショット読み取りフェーズ中にバックフィルをスキップしません。 バックフィルをスキップすると、スナップショットフェーズ中のテーブルへの変更は、スナップショットにマージされる代わりに、後の増分フェーズで読み取られます。 重要 バックフィルをスキップすると、スナップショットフェーズ中に発生した変更が再実行される可能性があるため、データの一貫性が損なわれる可能性があります。少なくとも 1 回のセマンティクスのみが保証されます。 説明 VVR 11.1 以降でのみサポートされます。
  scan.incremental.snapshot.unbounded-chunk-first.enabled	スナップショット読み取りフェーズ中に、境界のないチャンクを最初に分散するかどうかを指定します。	いいえ	BOOLEAN	false	有効な値： true：スナップショット読み取りフェーズ中に、境界のないチャンクを最初に分散します。 false (デフォルト)：スナップショット読み取りフェーズ中に、境界のないチャンクを最初に分散しません。 これは実験的な機能です。有効にすると、スナップショットフェーズ中に TaskManager が最後のチャンクを同期する際の Out-of-Memory (OOM) エラーのリスクを軽減できます。ジョブの最初の起動前に追加することを推奨します。 説明 VVR 11.1 以降でのみサポートされます。
  binlog.session.network.timeout	binlog 接続のネットワーク読み取り/書き込みタイムアウト。	いいえ	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 以降でのみサポートされます。

• ディメンションテーブル固有

  パラメーター	説明	必須	データの型	デフォルト値	備考
  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 の 3 つのキャッシュポリシーをサポートします。値の詳細については、「背景情報」をご参照ください。 説明 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 に設定する必要があります。そうしないと、ジョブが異常に実行される可能性があります。

• シンク固有の

  パラメーター	説明	必須	データの型	デフォルト値	備考
  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	1s	なし。
  sink.ignore-delete	DELETE 操作を無視するかどうかを指定します。	いいえ	BOOLEAN	false	Flink SQL によって生成されたストリームに削除または更新前レコードが含まれている場合、複数の出力タスクが同じテーブルの異なるフィールドを同時に更新すると、データの一貫性が損なわれる可能性があります。 たとえば、レコードが削除された後、別のタスクが一部のフィールドのみを更新します。更新されなかったフィールドは 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 は MySQL BLOB 型のレコードを最大 2,147,483,647 (2^31 - 1) バイトまでしかサポートしません。
  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

構成項目

型マッピング

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