DataWorks:ECS 上の自己管理 MySQL データベースから MaxCompute へのデータ移行

操作手順

1. DataWorks で MySQL データソースを作成します。

   1. [データソース] ページに移動します。

      1. DataWorks コンソールにログインします。 上部のナビゲーションバーで、目的のリージョンを選択します。 左側のナビゲーションウィンドウで、[その他] > [管理センター]を選択します。 表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[管理センターへ移動]をクリックします。

      2. SettingCenter ページの左側のナビゲーションウィンドウで、[Data Sources] をクリックします。

   2. [データソースの追加] をクリックします。

   3. 「[データソースの追加]」ダイアログボックスで、[MySQL] をクリックします。

   4. [MySQL データソースの作成] ダイアログボックスで、パラメーターを設定します。詳細については、「MySQL データソースを設定する」をご参照ください。

      このトピックでは、[接続文字列モード] を例として使用します。[接続アドレス] には、メモした ECS インスタンスのプライベート IP アドレスとデフォルトの MySQL ポート番号 3306 を入力します。

      説明

      VPC 内のセルフマネージド MySQL データソースでは、[接続性テスト]機能はサポートされていません。そのため、接続性テストが失敗するのは正常です。

   5. [接続のテスト] をリソースグループに対してクリックします。

      データ統合専用リソースグループを使用して、データ同期タスクを実行できます。タスクは、実行ごとに 1 つのリソースグループしか使用できません。複数のリソースグループが利用可能な場合は、使用したい各リソースグループの接続性をテストする必要があります。これにより、同期タスクで使用されるリソースグループがデータソースに接続できることが保証されます。そうしないと、データ同期タスクは期待どおりに実行できません。詳細については、「ネットワーク接続ソリューション」をご参照ください。

   6. 接続性テストが成功した後、[完了] をクリックします。

2. MaxCompute テーブルを作成します。

   DataWorks でテーブルを作成し、MySQL データベースからのテストデータを受け取ります。

   1. 左上隅の アイコンをクリックして、[すべての製品] > [データ開発とタスク操作] > [DataStudio] を選択します。

   2. [ワークフロー] を作成します。詳細については、「スケジュールされたワークフローの作成」をご参照ください。

   3. 新しいワークフローを右クリックし、[テーブルの作成] > [MaxCompute] > [テーブル] を選択します。

   4. MaxCompute テーブルの名前を入力します。この例では、MySQL データベースのテーブルと同じ名前の good_sale を使用します。[DDL] をクリックし、テーブル作成文を入力してから、[テーブルスキーマの生成] をクリックします。

      この例では、次のテーブル作成文を使用します。データの型の変換にご注意ください。

      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. テーブルの[表示名]を入力し、[本番環境にコミット]をクリックして MaxCompute に good_sale テーブルを作成します。

3. データ統合タスクを設定します。

   1. ワークフローを右クリックし、[ノードの作成] > [データ統合] > [オフライン同期] を選択してデータ統合タスクを作成します。

   2. ソースには、追加した MySQL データソースを選択します。送信先には、追加した MaxCompute データソースを選択します。[コードエディタに切り替える] をクリックして、データ統合タスクをコードエディタに切り替えます。

      エラーが発生する場合や、ソースの [テーブル] を選択できない場合でも、これは正常な現象です。直接コードエディタに切り替えてください。

   3. 右側のペインで、[データ統合リソースグループ設定] タブをクリックし、購入した専用リソースグループを選択します。

      タスクをデータ統合専用リソースグループに切り替えないと、タスクは実行できません。

   4. データ統合タスクのスクリプトに次の内容を入力します。

      {
          "type": "job",
          "steps": [
              {
                  "stepType": "mysql",
                  "parameter": {
                      "column": [// ソース列の名前。
                          "create_time",
                          "category",
                          "brand",
                          "buyer_id",
                          "trans_num",
                          "trans_amount",
                          "click_cnt"
                      ],
                      "connection": [
                          {
                              "datasource": "shuai",// ソースデータソース。
                              "table": [
                                  "good_sale"// ソースデータベースのテーブル名。名前は角括弧 [] で囲まれた配列形式である必要があります。
                              ]
                          }
                      ],
                      "where": "",
                      "splitPk": "",
                      "encoding": "UTF-8"
                  },
                  "name": "Reader",
                  "category": "reader"
              },
              {
                  "stepType": "odps",
                  "parameter": {
                      "partition": "",
                      "truncate": true,
                      "datasource": "odps_source",// 宛先 MaxCompute データソースの名前。
                      "column": [// 宛先列の名前。
                          "create_time",
                          "category",
                          "brand",
                          "buyer_id",
                          "trans_num",
                          "trans_amount",
                          "click_cnt"
                      ],
                      "emptyAsNull": false,
                      "table": "good_sale"// 宛先テーブルの名前。
                  },
                  "name": "Writer",
                  "category": "writer"
              }
          ],
          "version": "2.0",
          "order": {
              "hops": [
                  {
                      "from": "Reader",
                      "to": "Writer"
                  }
              ]
          },
          "setting": {
              "errorLimit": {
                  "record": "0"
              },
              "speed": {
                  "throttle": false,
                  "concurrent": 2
              }
          }
      }

   5. 「実行」をクリックします。データが MaxCompute に転送されたかどうかを確認するには、画面下部の [実行時ログ] を表示します。