すべてのプロダクト
Search

このプロダクト

    Realtime Compute for Apache Flink:Flink CDC：MySQL データベース全体を Kafka へ同期

    ドキュメントセンター

    Realtime Compute for Apache Flink:Flink CDC：MySQL データベース全体を Kafka へ同期

    最終更新日:Jun 22, 2026

    このトピックでは、MySQL データベース全体を Kafka へ同期する方法について説明します。このアプローチにより、複数のジョブが MySQL データベースに与える負荷を軽減できます。

    背景情報

    MySQL CDC ソーステーブルは、MySQL からデータをキャプチャし、テーブルのリアルタイムな変更を同期します。これは、テーブルが他のデータテーブルとの JOIN 演算でディメンションテーブルとして機能する場合など、複雑なコンピューティングシナリオで一般的です。単一の MySQL テーブルが複数のジョブの依存関係になる可能性があります。複数のジョブが同じ MySQL テーブルからデータを処理する場合、データベースが複数の接続を開くため、MySQL サーバーとネットワークに大きな負荷がかかります。

    仕組み

    アップストリームの MySQL データベースへの負荷を軽減するために、Realtime Compute for Apache Flink は MySQL データベース全体を Kafka へ同期できます。このソリューションでは、Kafka を中間層として導入し、Flink CDC データインジェストジョブを使用してデータを Kafka へ同期します。

    単一のジョブで、アップストリームの MySQL データベースからのデータが Kafka へリアルタイムで同期されます。各 MySQL テーブルは、アップサートモードで対応する Kafka トピックへ書き込まれます。その後、ダウンストリームのジョブは、MySQL テーブルへ直接アクセスする代わりに、アップサート Kafka コネクタを使用してトピックからデータを読み取ります。この方法により、複数のジョブが MySQL データベースに与える負荷を効果的に軽減できます。

    图片 1

    制限事項

    • 同期する各 MySQL テーブルは、プライマリキーを持つ必要があります。

    • セルフマネージド Kafka クラスター、EMR Kafka クラスター、またはApsaraMQ for Kafka を使用できます。ApsaraMQ for Kafka を使用する場合は、デフォルトエンドポイント経由でのみ接続できます。

    • Kafka クラスターのストレージ容量は、ソーステーブルのストレージ容量よりも大きくする必要があります。そうでない場合、ストレージ不足によりデータが失われる可能性があります。データベース同期用に作成されたトピックは圧縮トピックです。圧縮トピックでは、各メッセージキーの最新のメッセージのみが保持されますが、データが期限切れになることはありません。つまり、圧縮トピックには、ソーステーブルのサイズとほぼ同等のデータ量が格納されます。

    シナリオ例

    たとえば、リアルタイム注文レビュー分析シナリオでは、ユーザーテーブル (user)、注文テーブル (order)、およびユーザーフィードバックテーブル (feedback) の 3 つのテーブルがあると仮定します。これらのテーブルには、次の図に示すデータが含まれています。mysql database

    ユーザーの注文情報とレビューを表示するには、user テーブルを結合して、name フィールドからユーザー名を取得する必要があります。次の SQL サンプルは、この操作を示しています。

    -- 注文情報をユーザーテーブルと結合して、各注文のユーザー名と製品名を表示します。
SELECT order.id as order_id, product, user.name as user_name
FROM order LEFT JOIN user
ON order.user_id = user.id;
-- レビューをユーザーテーブルと結合して、各レビューの内容と対応するユーザー名を表示します。
SELECT feedback.id as feedback_id, comment, user.name as user_name
FROM feedback LEFT JOIN user
ON feedback.user_id = user.id;

    前述の 2 つの SQL ジョブは、どちらもuser テーブルを使用します。実行時、両方のジョブは MySQL からフル読み取りと増分読み取りを行います。フル読み取りには MySQL 接続の作成が、増分読み取りには Binlog クライアントの作成が必要です。ジョブの数が増えるにつれて、MySQL 接続と Binlog クライアントリソースの需要も増加し、アップストリームのデータベースに大きな負荷がかかります。この負荷を軽減するため、Flink CDC データインジェストジョブを使用してアップストリームの MySQL データベースから Kafka へデータをリアルタイムで同期し、複数のダウンストリームジョブで消費できるようにします。

    前提条件

    事前準備

    MySQL データソースの準備

    1. ApsaraDB RDS for MySQL データベースを作成します。詳細については、「データベースの作成」をご参照ください。

      対象インスタンスにorder_dw という名前のデータベースを作成します。

    2. MySQL CDC データソースを準備します。

      1. インスタンス詳細ページの上部にある[データベースにログオン]をクリックします。

      2. 表示される DMS ログオンダイアログボックスで、作成したデータベースアカウントのユーザー名とパスワードを入力し、[ログイン] をクリックします。

      3. ログイン後、左側のペインでorder_dw データベースをダブルクリックして、データベースを切り替えます。

      4. SQL コンソールで、3 つのビジネステーブルを作成する DDL ステートメントと、データを挿入するステートメントを入力します。

        CREATE TABLE `user` (
  id bigint not null primary key,
  name varchar(50) not null
);
CREATE TABLE `order` (
  id bigint not null primary key,
  product varchar(50) not null,
  user_id bigint not null
);
CREATE TABLE `feedback` (
  id bigint not null primary key,
  user_id bigint not null,
  comment varchar(50) not null
);
-- データを準備します
INSERT INTO `user` VALUES(1, 'Tom'),(2, 'Jerry');
INSERT INTO `order` VALUES
(1, 'Football', 2),
(2, 'Basket', 1);
INSERT INTO `feedback` VALUES
(1, 1, 'Good.'),
(2, 2, 'Very good');

    3. [実行] をクリックし、次に [実行] をクリックします。

    操作手順

    1. Flink CDC データインジェストジョブを作成して起動し、アップストリームの MySQL データベースから Kafka へデータをリアルタイムで同期して、複数のダウンストリームジョブで消費できるようにします。データベース同期ジョブは、トピックを自動的に作成します。route モジュールを使用して、トピック名を定義できます。トピックのパーティション数とレプリカ数には Kafka クラスターのデフォルト設定が使用され、cleanup.policy は compact に設定されます。

      デフォルトのトピック名

      デフォルトでは、データベース同期ジョブによって作成される Kafka トピックは、{database_name}.{table_name} という命名形式になります。次のジョブでは、order_dw.user、order_dw.order、およびorder_dw.feedback の 3 つのトピックが作成されます。

      1. [Development] > [Data Ingestion] ページで、Flink CDC データインジェストジョブを作成し、次のコードを YAML エディターにコピーします。

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{username}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (オプション) 増分フェーズ中に新しく作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (オプション) テーブルとフィールドのコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限スプリットを優先的にディスパッチし、潜在的なTaskManagerのOutOfMemory問題を回避します。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 解析フィルターを有効にして、読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # ApsaraMQ for Kafka には、次のパラメータが必要です。
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. 右上隅で、[デプロイ] をクリックしてジョブをデプロイします。

      3. 左側のナビゲーションバーで、[オペレーションセンター] > [デプロイ] をクリックします。対象ジョブの [アクション] 列で [開始] をクリックし、[初期モード] を選択して、[開始] をクリックします。

      テーブルごとのトピック名

      route モジュールを使用して、各テーブルのトピック名を指定できます。次のジョブでは、user1、order2、およびfeedback3 の 3 つのトピックが作成されます。

      1. [Development] > [Data Ingestion] ページで、Flink CDC データインジェストジョブを作成し、次のコードを YAML エディターにコピーします。

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{username}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (オプション) 増分フェーズ中に新しく作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (オプション) テーブルとフィールドのコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限スプリットを優先的にディスパッチし、潜在的なTaskManagerのOutOfMemory問題を回避します。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 解析フィルターを有効にして、読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  - source-table: order_dw.user
    sink-table: user1
  - source-table: order_dw.order
    sink-table: order2
  - source-table: order_dw.feedback
    sink-table: feedback3
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # ApsaraMQ for Kafka には、次のパラメータが必要です。
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. 右上隅で、[デプロイ]をクリックしてジョブをデプロイします。

      3. 左側のナビゲーションバーで、[運用センター] > [デプロイ] を選択し、対象のジョブの [アクション] 列の [開始] をクリックし、[初期モード] を選択してから、[開始] をクリックします。

      バッチトピック名

      route モジュールを使用して、生成されるトピック名のパターンを指定できます。次のジョブでは、topic_user、topic_order、およびtopic_feedback の 3 つのトピックが作成されます。

      1. [Development] > [Data Ingestion] ページで、Flink CDC データインジェストジョブを作成し、次のコードを YAML エディターにコピーします。

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{username}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (オプション) 増分フェーズ中に新しく作成されたテーブルからデータを同期します。
  scan.binlog.newly-added-table.enabled: true
  # (オプション) テーブルとフィールドのコメントを同期します。
  include-comments.enabled: true
  # (オプション) 無制限スプリットを優先的にディスパッチし、潜在的なTaskManagerのOutOfMemory問題を回避します。
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (オプション) 解析フィルターを有効にして、読み取りを高速化します。
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  - source-table: order_dw.\.*
    sink-table: topic_<>
    replace-symbol: <>
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # ApsaraMQ for Kafka には、次のパラメータが必要です。
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. 右上隅で[デプロイ]をクリックしてジョブをデプロイします。

      3. 左側のナビゲーションバーで、[オペレーションセンター] > [デプロイ] をクリックします。対象のジョブの [アクション] 列で [開始] をクリックし、[初期モード] を選択して、[開始] をクリックします。

    1. Kafka データをリアルタイムで消費します。

      データインジェストジョブは、アップストリームの MySQL データベースからのデータを JSON 形式で Kafka へ書き込みます。その後、複数のダウンストリームジョブが単一のトピックからデータを消費して、データベーステーブルの最新の状態を取得できます。Kafka へ同期されたテーブルからデータを消費するには、次のいずれかの方法を使用できます。

      カタログによる方法

      Kafka トピックをソーステーブルとして使用して、データを読み取ります。

      1. [Development] > [ETL] ページで、ストリーミング SQL ジョブを作成し、次のコードを SQL エディターにコピーします。

        CREATE TEMPORARY TABLE print_user_proudct(
  order_id BIGINT,
  product STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
CREATE TEMPORARY TABLE print_user_feedback(
  feedback_id BIGINT,
  `comment` STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
BEGIN STATEMENT SET;      -- 複数のシンクへ書き込む場合に必要です。
-- 注文情報を Kafka JSON カタログのユーザーテーブルと結合して、各注文のユーザー名と製品名を表示します。
INSERT INTO print_user_proudct
SELECT `order`.key_id as order_id, value_product as product, `user`.value_name as user_name
FROM `kafka-catalog`.`kafka`.`order`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/  as `order` -- グループと起動モードを指定します。
LEFT JOIN `kafka-catalog`.`kafka`.`user`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/ as `user` -- グループと起動モードを指定します。
ON `order`.value_user_id = `user`.key_id;
-- レビューをユーザーテーブルと結合して、各レビューの内容と対応するユーザー名を表示します。
INSERT INTO print_user_feedback
SELECT feedback.key_id as feedback_id, value_comment as `comment`, `user`.value_name as user_name
FROM `kafka-catalog`.`kafka`.feedback/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/  as feedback  -- グループと起動モードを指定します。
LEFT JOIN `kafka-catalog`.`kafka`.`user`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/ as `user` -- グループと起動モードを指定します。
ON feedback.value_user_id = `user`.key_id;
END;      -- 複数のシンクへ書き込む場合に必要です。

        この例では、Print コネクタを使用して結果を直接出力しています。さらに分析するために、別のコネクタを使用する結果テーブルに結果を出力することもできます。複数のシンクへ書き込む構文の詳細については、「INSERT INTO ステートメント」をご参照ください。

        説明

        この方法を直接使用すると、スキーマの変更が発生する可能性があります。その結果、Kafka JSON カタログによって解析されるスキーマが、対応する MySQL テーブルのスキーマと異なる場合があります。たとえば、削除されたフィールドが引き続き表示されたり、一部のフィールドに null 値が含まれたりする可能性があります。

        カタログから読み取られるスキーマは、消費されたデータのフィールドで構成されます。フィールドが削除されても、そのメッセージの有効期限が切れていない場合、フィールドが null 値で表示される可能性があります。この場合、特別な処理は必要ありません。

      2. 右上隅で [デプロイ] をクリックしてジョブをデプロイします。

      3. 左側のナビゲーションバーで [O&M センター] > [デプロイ] の順にクリックし、対象のジョブの [アクション] 列で [開始] をクリックし、[初期モード] を選択してから、[開始] をクリックします。

      一時テーブルによる方法

      カスタムスキーマを定義し、一時テーブルからデータを読み取ります。

      1. [Development] > [ETL] ページで、ストリーミング SQL ジョブを作成し、次のコードを SQL エディターにコピーします。

        CREATE TEMPORARY TABLE user_source (
  key_id BIGINT,
  value_name STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'user',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE order_source (
  key_id  BIGINT,
  value_product STRING,
  value_user_id BIGINT  
) WITH (
  'connector' = 'kafka',
  'topic' = 'order',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE feedback_source (
  key_id  BIGINT,
  value_user_id BIGINT,
  value_comment STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'feedback',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE print_user_proudct(
  order_id BIGINT,
  product STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
CREATE TEMPORARY TABLE print_user_feedback(
  feedback_id BIGINT,
  `comment` STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
BEGIN STATEMENT SET;      -- 複数のシンクへ書き込む場合に必要です。
-- 注文情報をユーザーテーブルと結合して、各注文のユーザー名と製品名を表示します。
INSERT INTO print_user_proudct
SELECT order_source.key_id as order_id, value_product as product, user_source.value_name as user_name
FROM order_source LEFT JOIN user_source
ON order_source.value_user_id = user_source.key_id;
-- レビューをユーザーテーブルと結合して、各レビューの内容と対応するユーザー名を表示します。
INSERT INTO print_user_feedback
SELECT feedback_source.key_id as feedback_id, value_comment as `comment`, user_source.value_name as user_name
FROM feedback_source  LEFT JOIN user_source
ON feedback_source.value_user_id = user_source.key_id;
END;      -- 複数のシンクへ書き込む場合に必要です。

        この例では、Print コネクタを使用して結果を直接出力しています。さらに分析するために、別のコネクタの結果テーブルに結果を出力することもできます。複数のシンクへ書き込む構文の詳細については、「INSERT INTO ステートメント」をご参照ください。

        次の表に、一時テーブルの構成パラメータを示します。

        パラメータ

        説明

        備考

        connector

        コネクタのタイプ。

        値を kafka に設定します。

        topic

        対応するトピックの名前。

        Kafka JSON カタログの記述と一致している必要があります。

        properties.bootstrap.servers

        Kafka ブローカーのアドレス。

        形式はhost:port,host:port,host:portで、カンマ (,) で区切ります。

        scan.startup.mode

        Kafka からデータを読み取る開始位置。

        有効な値：

        • earliest-offset：利用可能な最も古いオフセットから読み取りを開始します。

        • latest-offset：最新のオフセットから読み取りを開始します。

        • group-offsets (デフォルト)：properties.group.id で指定されたグループのコミット済みオフセットから読み取ります。

        • timestamp：scan.startup.timestamp-millis で指定されたタイムスタンプから読み取ります。

        • specific-offsets：scan.startup.specific-offsets で指定されたオフセットから読み取りを開始します。

        注意

        このパラメータは、ジョブが保存された状態なしで起動する場合に有効になります。ジョブが再起動するか、チェックポイントから復旧する場合、保存された状態からの読み取りが優先されます。

        key.format

        Kafka メッセージキーのシリアル化/逆シリアル化に使うフォーマット。

        値を json に設定します。

        key.fields

        Kafka メッセージキーに対応する、ソーステーブルまたは結果テーブルのフィールド。

        複数のフィールド名を区切るには、セミコロン (;) を使用します。例：field1;field2。

        key.fields-prefix

        すべての Kafka メッセージキーフィールドに適用されるカスタムプレフィックス。メッセージバリューフィールドやメタデータフィールドとの名前の競合を回避するために使用します。

        この値は、Kafka JSON カタログのkey.fields-prefix パラメータの値と一致させる必要があります。

        value.format

        Kafka メッセージバリューのシリアル化/逆シリアル化に使うフォーマット。

        値を json に設定します。

        value.fields-prefix

        すべての Kafka メッセージバリューフィールドに適用されるカスタムプレフィックス。メッセージキーフィールドやメタデータフィールドとの名前の競合を回避するために使用します。

        Kafka JSON カタログのvalue.fields-prefix パラメータの値と一致させる必要があります。

        value.fields-include

        メッセージバリュー内でのメッセージキーフィールドの処理ポリシー。

        値を EXCEPT_KEY に設定します。これは、メッセージバリューにメッセージキーのフィールドが含まれないことを示します。

        value.json.infer-schema.flatten-nested-columns.enable

        Kafka メッセージバリュー内のネストされた JSON 列を再帰的に展開するかどうかを指定します。

        対応するカタログのinfer-schema.flatten-nested-columns.enable パラメータの値と一致させる必要があります。

        value.json.infer-schema.primitive-as-string

        Kafka メッセージバリュー内のすべてのプリミティブ型を String として推論するかどうかを指定します。

        対応するカタログのinfer-schema.primitive-as-string パラメータの値と一致させる必要があります。

      2. 右上隅の[デプロイ]をクリックしてジョブをデプロイします。

      3. 左側のナビゲーションバーで [オペレーションセンター] > [デプロイメント] の順にクリックし、対象のジョブの [操作] 列で [開始] をクリックし、[初期モード] を選択してから、[開始] をクリックします。

    2. ジョブの結果を表示します。

      1. 左側のナビゲーションバーで、[Operations Center] > [Deployments] をクリックし、対象のジョブをクリックします。

      2. [ジョブログ] タブの [実行中の TaskManager] タブで、表示したい [パス、ID] を持つタスクをクリックします。

      3. [ログ] をクリックし、 PrintSinkOutputWriter に関連するログ情報を検索します。

        ログで PrintSinkOutputWriter の出力を検索します。出力には、+I[1, Good., Tom]、+I[2, Very good, Jerry]、+I[2, Basket, Tom]、および+I[1, Football, Jerry] という 4 つの結合済みデータレコードが含まれています。これは、ユーザーテーブルと注文およびフィードバックテーブルの間の結合が成功したことを示しています。

    関連ドキュメント