ECS インスタンス上のユーザー作成 MySQL データベースの MaxCompute へのデータ移行 -

ここでは、データ統合に排他的リソースグループを使用して、Elastic Compute Service (ECS) インスタンスでユーザーが作成した MySQL データベースから MaxCompute にデータを移行する方法について説明します。

始める前に

ECS インスタンスが購入され、Virtual Private Cloud (VPC) にバインドされていること (クラシックネットワークではなく)。 MySQL データベースが ECS インスタンスにデプロイされていること。データベースに接続するためのアカウントが作成され、データベースにはテストデータが含まれていること。ここでは、次の SQL 文を使用して MySQL データベースのテーブルを作成し、テーブルにテストデータを挿入します。

CREATE TABLE IF NOT EXISTS good_sale(
   create_time timestamp,
   category varchar(20),
   brand varchar(20),
   buyer_id varchar(20),
   trans_num varchar(20),
   trans_amount DOUBLE,
   click_cnt varchar(20)
   );
insert into good_sale values('2018-08-21','coat','brandA','lilei',3,500.6,7),
('2018-08-22','food','brandB','lilei',1,303,8),
('2018-08-22','coat','brandC','hanmeimei',2,510,2),
('2018-08-22','bath','brandA','hanmeimei',1,442.5,1),
('2018-08-22','food','brandD','hanmeimei',2,234,3),
('2018-08-23','coat','brandB','jimmy',9,2000,7),
('2018-08-23','food','brandA','jimmy',5,45.1,5),
('2018-08-23','coat','brandE','jimmy',5,100.2,4),
('2018-08-24','food','brandG','peiqi',10,5560,7),
('2018-08-24','bath','brandF','peiqi',1,445.6,2),
('2018-08-24','coat','brandA','ray',3,777,3),
('2018-08-24','bath','brandG','ray',3,122,3),
('2018-08-24','coat','brandC','ray',1,62,7) ;

ECS インスタンスのプライベート IP アドレス、VPC、および VSwitch が記録されていること。
ECS インスタンスにセキュリティグループルールが追加され、MySQL データベースが使用するポートでのアクセスリクエストが許可されていること。 MySQL データベースはデフォルトでポート番号 3306 を使用します。詳細については、「セキュリティグループルールを追加」をご参照ください。セキュリティグループの名前が記録されていること。
ECS インスタンスと同じリージョンに属する DataWorks ワークスペースが作成されていること。この例では、基本モードにあり、計算エンジンとして MaxCompute を使用するワークスペースを作成します。詳細については、「ワークスペースの作成」をご参照ください。
データ統合用の排他的リソースグループが購入され、ECS インスタンスが存在する VPC にバインドされていること。排他的リソースグループと ECS インスタンスは同じゾーンにあります。詳細については、「Use exclusive resource groups for data integration」をご参照ください。排他的リソースグループが VPC にバインドされた後は、排他的リソースグループに関する情報をページで見ることができます。
排他的リソースグループの VPC、VSwitch、およびセキュリティグループは、ECS インスタンスのものと同じであること。 VPC、VSwitch、および排他的リソースグループのセキュリティグループを表示するには、次の手順に従います。[リソースグループ] ページに移動し、[排他的リソースグループ] タブで専用リソースグループを検索します。 [操作] 列で、[VPC バインディングの追加] をクリックします。表示されるページで、排他的リソースグループのVPC、VSwitch、およびセキュリティグループを表示できます。

このタスクについて

排他的リソースグループは、高速かつ安定した方法でデータを送信できます。データ統合用の排他的リソースグループは、データストアと同じゾーンにあり、対応するワークスペースと同じリージョンに属している場合にのみ、データストアに適切にアクセスできます。この例では、ターゲットデータストアは、ECS インスタンス上でユーザーが作成した MySQL データベースです。

手順

DataWorks コンソールで MySQL データベースへの接続を作成します。
1. Alibaba Cloud アカウントを使用して DataWorks コンソールへログインします。
2. 左側のナビゲーションペインで、[ワークスペース] をクリックします。表示される [ワークスペース] ページで、ターゲットワークスペースを探し、[操作] 列の [データ統合] をクリックします。
3. 表示されるページでの左側のナビゲーションペインで [接続] をクリックします。
4. 表示される [データソース] ページで、右上隅に表示される [接続の追加] をクリックします。
5. 表示される [接続の追加] ダイアログボックスで、[リレーショナルデータベース] セクションにある [MySQL] を選択します。
6. 表示される [MySQL 接続の追加] ダイアログボックスで、必要に応じてパラメーターを設定します。
  この例では、[接続先] を [接続モード] に設定し、[JDBC URL] に jdbc:mysql://ServerIP:Port/Database 形式の JDBC URL を設定します。形式にある ServerIP および Port を、それぞれ ECS インスタンスのプライベート IP アドレスおよび MySQL データベースのデフォルトのポート番号 3306 に置き換えます。 MySQL データベース用に作成したユーザー名とパスワードを入力し、 [完了] をクリックします。
  
  注現在、DataWorks は VPC でユーザーが作成した MySQL データベースの接続をテストできません。したがって、接続テストは失敗しますが、これは正常です。
MaxCompute テーブルを作成します。
MySQL データベースからテストデータを受信するには、DataWorks コンソールで MaxCompute テーブルを作成する必要があります。
1. 左上隅の DataWorks アイコンをクリックして、 [すべてのプロダクト] > [DataStudio] を選択します。
2. 表示されるページで、ワークフローを作成します。
3. 作成したワークフローを右クリックし、[作成] > [MaxCompute][テーブル] を選択します。
4. 表示される [テーブルの作成] ダイアログボックスで、[エンジンタイプを選択してください] および [テーブル名] を設定し、[確定] をクリックします。テーブルが作成されます。この例では、テーブル名を good_sale に設定します。これは、MySQL データベースのテーブルと同じ名前です。テーブルの [設定] タブで [DDL 文] をクリックし、テーブル作成文を入力して、[テーブルスキーマの生成] をクリックします。
  この例では、次の SQL 文を入力します。データ型変換に注意してください。
```
CREATE TABLE IF NOT EXISTS good_sale(
   create_time string,
   category STRING,
   brand STRING,
   buyer_id STRING,
   trans_num BIGINT,
   trans_amount DOUBLE,
   click_cnt BIGINT
   );
```
5. [表示名] を設定し、[本番環境への送信] をクリックします。 good_sale という名前の MaxCompute テーブルが送信されます。

バッチ同期ノードを設定します。

作成したワークフローを右クリックし、[作成] > [データ統合] > [バッチ同期] を選択し、バッチ同期ノードを作成します。
バッチ同期ノードの [設定] タブで、[ソース] の下の [接続] を作成した MySQL 接続に、[ターゲット] の下の [接続] を ODPS odps_first に設定します。 [コードエディターに切り替え]アイコンをクリックしてコードエディターに切り替えます。
[ソース] の下にある [テーブル] を設定できない場合、またはコードエディターに切り替えようとしたときにエラーが返された場合は、問題を無視してください。
右側のナビゲーションペインで、[リソースグループの設定] をクリックします。表示される [リソースグループ] 設定ペインで、[プログラム] を [排他的リソースグループ] に設定し、購入した排他的リソースグループを選択します。
リソースグループをデータ統合用の排他的リソースグループに変更しない場合、同期ノードは実行されません。

バッチ同期ノードのコードを入力します。この例では、次のコードを入力します。

{
    "type": "job",
    "steps": [
        {
            "stepType": "mysql",
            "parameter": {
                "column": [// The columns in the source table.
                    "create_time",
                    "category",
                    "brand",
                    "buyer_id",
                    "trans_num",
                    "trans_amount",
                    "click_cnt"
                ],
                "connection": [
                    {
                        "datasource": "shuai",// The name of the source data store.
                        "table": [
                            "good_sale"// The name of the table in the source data store. The name must be enclosed in brackets ([ ]).
                        ]
                    }
                ],
                "where": "",
                "splitPk": "",
                "encoding": "UTF-8"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "odps",
            "parameter": {
                "partition": "",
                "truncate": true,
                "datasource": "odps_first",// The name of the destination data store.
                "column": [// The columns in the destination table.
                    "create_time",
                    "category",
                    "brand",
                    "buyer_id",
                    "trans_num",
                    "trans_amount",
                    "click_cnt"
                ],
                "emptyAsNull": false,
                "table": "good_sale"// The name of the destination table.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle": false,
            "concurrent": 2
        }
    }
}

[実行] アイコンをクリックします。バッチ同期ノードの操作ログは、 [実行ログ] タブで表示できます。

タスクの結果

テストデータが MaxCompute テーブルに同期されているかどうかを確認するには、ODPS SQL ノードを作成します。ODPS SQL ノードの [設定] タブで、SQL 文 select * from good_sale; を入力し、[実行] アイコンをクリックします。テストデータが表示される場合、MaxCompute テーブルに同期されています。