すべてのプロダクト
Search

このプロダクト

    DataHub:RDS、MySQL、AnalyticDB とのデータ同期

    ドキュメントセンター

    DataHub:RDS、MySQL、AnalyticDB とのデータ同期

    最終更新日:Jun 16, 2026

    RDS へのデータ同期

    前提条件

    1. RDS インスタンスとテーブル構造の準備 RDS コンソールで RDS インスタンスを作成します。クラシックネットワークまたは VPC ネットワークのいずれかを使用できます。DataHub 同期タスクを作成する際に、正しいネットワークタイプを選択してください。同期中、DataHub はフィールドのデータ型を、以下の表のように対応する RDS のデータ型にマッピングします。

    DataHub

    RDS

    TINYINT

    TINYINT

    SMALLINT

    SMALLINT

    INTEGER

    INT

    BIGINT

    BIGINT

    STRING

    VARCHAR

    BOOLEAN

    BOOLEAN / TINYINT

    FLOAT

    FLOAT

    DOUBLE

    DOUBLE

    TIMESTAMP

    TIMESTAMP / BIGINT

    DECIMAL

    DECIMAL

    DataHub の TINYINT、SMALLINT、INTEGER、および FLOAT データ型は、Java SDK バージョン 2.16.1-public 以降でサポートされています。 2. RDS IP ホワイトリストと内部アドレスの設定 DataHub の同期機能を使用するには、RDS コンソールで DataHub サービスの IP アドレスを IP ホワイトリストに追加する必要があります。 これにより、DataHub はお使いの RDS サービスにアクセスできるようになります。 必要な DataHub サービスの IP アドレスについては、「概要」をご参照ください。 また、同期タスクを作成する際には、RDS の内部アドレスを指定する必要もあります。

    3. 同期に関する注意事項

    1. DataHub は TUPLE 型のデータのみを RDS に同期できます。

    2. DataHub の TIMESTAMP 型は マイクロ秒 をベースとしており、デフォルトで RDS TIMESTAMP 型に変換されます。データ精度にご注意ください。

    3. 同じプライマリキーに対する読み書きの同時実行によるロック競合を防ぐために、同じプライマリキーを持つデータが同じ DataHub シャードに書き込まれるようにしてください。

    4. VPC ネットワークを使用する場合、DataHub のトピックと RDS インスタンスが同じリージョンにあることを確認してください。

    同期タスクの作成

    1. プロジェクトリスト > プロジェクト詳細 > トピック詳細 ページに移動します。

    2. 右上隅にある+ 同期 ボタンをクリックして、同期タスクを作成します。 表示される [コネクタの作成] ダイアログボックスで、[ホスト]、[ポート] (デフォルトでは 3306)、[データベース]、[テーブル]、[ユーザー]、[パスワード] などの接続情報、および [書き込みモード]、[インポートフィールド]、[ネットワークタイプ] などの同期パラメーターを設定します。 VPC ネットワークを選択した場合は、[VPC ID] と [インスタンス ID] も指定する必要があります。 [開始時刻] と [タイムスタンプ単位] (デフォルトではマイクロ秒) を設定し、[作成] をクリックします。

    以下は、主要なコンソール設定です。高度なオプションについては、SDK ドキュメントをご参照ください。

    1. [ホスト] には、RDS の 内部アドレス を指定します。

    2. [インポートフィールド]：DataHub から RDS テーブルに同期する特定の列を選択できます。

    3. [書き込みモード]：IGNORE モードと OVERWRITE モードがあります。

      • IGNORE: 重複データをスキップし、INSERT IGNORE INTO の書き込みに対応します。

      • 上書き: 重複データを置き換えます。REPLACE INTO の書き込みに相当します。

    4. [VPC ID] と [インスタンス ID]：RDS インスタンスが VPC ネットワーク内にある場合、対応する VPC ID と RDS インスタンス ID を提供する必要があります。

      同期の例

    5. RDS インスタンスとテーブル構造を作成します。 RDS コンソールで、RDS インスタンスとデータテーブルを作成します。 この例では、テーブル名は wurao_tuple_test_001 で、field1 (varchar、長さ 255)、field2 (bigint、長さ 20)、field3 (double) の 3 つの列があります。

    6. DataHub トピックの作成 DataHub で、タイプが TUPLE のトピックを作成します。この例では、スキーマには field1 (STRING)、field2 (BIGINT)、および field3 (DOUBLE) の 3 つのフィールドがあります。どのフィールドも null にはできません。

    7. 同期タスクを作成します。[書き込みモード] を IGNORE に設定し、すべてのフィールド (field1、field2、field3) をインポートします。[ネットワークタイプ] として CLASSIC を選択し、[作成] をクリックします。

    8. DataHub に TUPLE データを書き込みます。 合計 4 件のレコードを書き込みます。 データは 2 件の一意のレコードで構成されており、それぞれが 2 回書き込まれます。 field1 の値は f_1 と f_2、field2 の値は 1 と 2、field3 の値は 1.0 と 2.0 です。

    9. 同期されたデータの検証。MySQL クライアントを使用して RDS サービスに接続し、テーブルをクエリします。結果は以下のとおりです。

      mysql> select * from wurao_tuple_test_001;
+--------+--------+--------+
| field1 | field2 | field3 |
+--------+--------+--------+
| f_1    |      1 |   1.00 |
| f_2    |      2 |   2.00 |
| f_1    |      1 |   1.00 |
| f_2    |      2 |   2.00 |
+--------+--------+--------+
4 rows in set (0.00 sec)

    MySQL へのデータ同期

    前提条件

    1. MySQL インスタンスとテーブル構造の準備 MySQL コンソールで MySQL インスタンスを作成します。

    同期中、DataHub はフィールドのデータ型を、以下の表のように対応する MySQL のデータ型にマッピングします。

    DataHub

    MySQL

    TINYINT

    TINYINT

    SMALLINT

    SMALLINT

    INTEGER

    INT

    BIGINT

    BIGINT

    STRING

    VARCHAR

    BOOLEAN

    BOOLEAN / TINYINT

    FLOAT

    FLOAT

    DOUBLE

    DOUBLE

    TIMESTAMP

    TIMESTAMP / BIGINT

    DECIMAL

    DECIMAL

    DataHub の TINYINT、SMALLINT、INTEGER、FLOAT データ型は、Java SDK バージョン 2.16.1-public 以降でサポートされています。

    2. 同期に関する注意事項 1) DataHub は、TUPLE データのみを MySQL サービスに同期できます。 2) デフォルトでは、DataHub の TIMESTAMP 値は、microseconds に基づいて MySQL TIMESTAMP 型に変換されます。データ精度を適宜管理してください。 3) 同じプライマリキーに対する同時読み取りおよび書き込みによるロック競合を回避するには、同じプライマリキーを持つデータが DataHub の同じシャードに書き込まれるようにしてください。

    同期タスクの作成

    1. プロジェクトリスト > プロジェクト詳細 > トピック詳細 ページに移動します。

    2. 右上隅にある[+ 同期]ボタンをクリックして、同期タスクを作成します。

    [新しいコネクタ] ダイアログボックスで、[ホスト] (MySQL 内部アドレス)、[ポート] (デフォルトは 3306)、[データベース]、[テーブル]、[ユーザー]、[パスワード] などの接続情報を設定します。 次に、[書き込みモード]、[インポートされるフィールド]、[ネットワークタイプ] (VPC ネットワークまたはクラシックネットワーク) などの同期パラメーターを設定します。 VPC ネットワークを選択した場合は、MySQL サーバーの [VPC ID] と [インスタンス ID] も指定する必要があります。 設定が完了したら、[作成] をクリックします。

    以下は、主要なコンソール設定です。高度なオプションについては、SDK ドキュメントをご参照ください。

    1. [ホスト] には、MySQL の 内部アドレス を指定します。

    2. [インポートフィールド]：DataHub から MySQL テーブルに同期する特定の列を選択できます。

    3. [書き込みモード]：IGNORE モードと OVERWRITE モードがあります。

      • IGNORE： INSERT IGNORE INTO を使用してデータを書き込み、重複データをスキップします。

      • OVERWRITE： REPLACE INTO 書き込み操作によって重複データを置換します。

    同期の例

    1. MySQL インスタンスとテーブル構造の作成。MySQL コンソールで MySQL インスタンスとデータテーブルを作成します。

    2. DataHub トピックの作成。DataHub で TUPLE 型のトピックを作成します。スキーマは以下のとおりです。

    この例では、トピック wurao_tuple_test_001 のスキーマには field1 (STRING)、field2 (BIGINT)、field3 (DOUBLE) の 3 つのフィールドがあります。どのフィールドも NULL 値を許容しません。

    1. 同期タスクの作成。[書き込みモード] を IGNORE に設定し、すべてのフィールドをインポートします。

    1. DataHub への TUPLE データの書き込み。合計 4 つのレコードを書き込みます。

    データは 2 つの一意のレコードで構成されており、それぞれ 2 回ずつ書き込まれています。 field1 の値は f_1 と f_2、field2 の値は 1 と 2、field3 の値は 1.0 と 2.0 です。

    1. 同期されたデータの検証。MySQL クライアントを使用して MySQL サービスに接続し、テーブルをクエリします。結果は以下のとおりです。

    mysql> select * from  wurao_tuple_test_001;
+--------+--------+--------+
| field1 | field2 | field3 |
+--------+--------+--------+
| f_1    |      1 |   1.00 |
| f_2    |      2 |   2.00 |
| f_1    |      1 |   1.00 |
| f_2    |      2 |   2.00 |
+--------+--------+--------+
4 rows in set (0.00 sec)

    AnalyticDB for MySQL 3.0 へのデータ同期

    前提条件

    1. AnalyticDB for MySQL インスタンスとテーブル構造の準備 コンソールで AnalyticDB for MySQL インスタンスを作成します。クラシックネットワークまたは VPC ネットワークのいずれかを使用できます。DataHub 同期タスクを作成する際に、正しいネットワークタイプを選択してください。同期中、DataHub はフィールドのデータ型を、以下の表のように対応する AnalyticDB for MySQL のデータ型にマッピングします。

    DataHub

    AnalyticDB

    TINYINT

    TINYINT

    SMALLINT

    SMALLINT

    INTEGER

    INT

    BIGINT

    BIGINT

    STRING

    VARCHAR

    BOOLEAN

    BOOLEAN / TINYINT

    FLOAT

    FLOAT

    DOUBLE

    DOUBLE

    TIMESTAMP

    TIMESTAMP / BIGINT

    DECIMAL

    DECIMAL

    DataHub の TINYINT、SMALLINT、INTEGER、FLOAT データ型は、Java SDK 2.16.1-public 以降でサポートされています。 2. ADS IP ホワイトリストと内部アドレス DataHub の同期機能を使用するには、RDS コンソールで IP ホワイトリストを設定して、DataHub が ADS サービスにアクセスできるようにする必要があります。 DataHub サービスの IP ホワイトリストについては、よくある質問をご参照ください。 また、同期タスクの作成時に ADS の内部アドレスを指定する必要があります。 3. 同期に関する注意点 1) DataHub は、TUPLE データのみを ADS に同期できます。 2) デフォルトでは、DataHub の TIMESTAMP データ型は、microsecond の精度で ADS TIMESTAMP 型に変換されます。 これに応じてデータ精度を管理してください。 3) 同一プライマリーキーのデータに対する同時読み書き操作によるロック競合を防ぐには、そのデータが同じ DataHub Shard に書き込まれるようにしてください。 4) VPC ネットワークを使用する場合は、DataHub トピックと ADS インスタンスが同じリージョンにあることを確認してください。

    同期タスクの作成

    1. プロジェクトリスト > プロジェクト詳細 > トピック詳細 ページに移動します。

    2. 右上隅にある [+ Sync] ボタンをクリックして、同期タスクを作成します。[コネクタの作成] ダイアログボックスが表示されます。ダイアログボックスで、[ホスト]、[ポート] (デフォルト: 3306)、[データベース]、[テーブル]、[ユーザー]、[パスワード] などの接続情報、および [書き込みモード]、[インポートするフィールド]、[ネットワークタイプ] などの同期パラメーターを設定します。VPC ネットワークタイプを選択した場合は、[VPC ID] と [インスタンス ID] も入力する必要があります。設定が完了したら、[作成] をクリックします。

    以下は、主要なコンソール設定です。高度なオプションについては、SDK ドキュメントをご参照ください。

    1. [ホスト]には、ADS の 内部アドレス を指定します。

    2. [インポートフィールド]：DataHub から AnalyticDB for MySQL テーブルに同期する特定の列を選択できます。

    3. [書き込みモード]：IGNORE モードと OVERWRITE モードがあります。

      • IGNORE: INSERT IGNORE INTO による書き込みで、重複データをスキップします。

      • OVERWRITE: 重複するデータを REPLACE INTO 書き込みで上書きします。

    4. インスタンス ID の取得方法

      1. 「クラスターの詳細属性のクエリ」をクリックします。

      2. [デバッグ] をクリックします。デバッグページで、サービスエンドポイントを選択し、[DBClusterId] (AnalyticDB for MySQL インスタンスのクラスター ID) を入力して、[呼び出しを開始] をクリックします。

        1. DescribeDBClusterAttribute API リファレンスページに移動します。このページには、API の説明と、右側に [デバッグ] ボタンがあります。

        2. [DBClusterId] に加えて、[RegionId] フィールドにクラスターの リージョン ID (例: cn-beijing) も入力する必要があります。 次に、[呼び出しの開始] をクリックします。

      3. レスポンス内の VPCCloudInstanceId フィールドの値が、必要なインスタンス ID です。

        1. ステータスコード 200 が返されてコールが成功した場合、返された JSON 結果でキーワード vpccloud を検索して VPCCloudInstanceId フィールドを探します。

    同期の例

    1. AnalyticDB for MySQL インスタンスとテーブル構造の作成。

    2. DataHub トピックの作成 DataHub で、TUPLE 型のトピックを作成します。この例では、スキーマには field1 (STRING)、field2 (BIGINT)、および field3 (DOUBLE) の 3 つのフィールドがあります。いずれのフィールドも null にすることはできません。

    3. 同期タスクを作成します。 [書き込みモード] を IGNORE に設定し、すべてのフィールド field1、field2、field3 をインポートします。 [ネットワークタイプ] として CLASSIC を選択し、[作成] をクリックします。

    4. DataHub に TUPLE データを書き込みます。合計 4 件のレコードを書き込みます。データは 2 つの一意のレコードで構成され、それぞれ 2 回書き込まれます。field1 の値は f_1 と f_2、field2 の値は 1 と 2、field3 の値は 1.0 と 2.0 です。

    5. 同期されたデータの検証。MySQL クライアントを使用して AnalyticDB for MySQL サービスに接続し、テーブルをクエリします。結果は以下のとおりです。

      mysql> select * from  wurao_tuple_test_001;
+--------+--------+--------+
| field1 | field2 | field3 |
+--------+--------+--------+
| f_1    |      1 |   1.00 |
| f_2    |      2 |   2.00 |
| f_1    |      1 |   1.00 |
| f_2    |      2 |   2.00 |
+--------+--------+--------+
4 rows in set (0.00 sec)