MaxCompute:ElasticsearchからMaxComputeへのデータ移行

手順

1. Elasticsearchでソーステーブルを作成します。 詳細については、「DataWorksを使用したMaxComputeからElasticsearchへのデータ同期」をご参照ください。

2. MaxComputeでターゲットテーブルを作成します。

   1. DataWorks コンソールにログインします。 上部のナビゲーションバーで、目的のリージョンを選択します。 左側のナビゲーションウィンドウで、[データ开発と操作とメンテナンス] > [データ開発] を選択します。 表示されるページで、ドロップダウンリストから目的のワークスペースを選択し、[入力データ開発] をクリックします。

   2. [DataStudio] ページの [スケジュール済みワークフロー] ペインで、作成したワークフローを見つけ、ワークフロー名を右クリックして、[テーブルの作成] > [MaxCompute] > [テーブル] を選択します。

   3. では、テーブルの作成ダイアログボックス, configure名前をクリックし、作成. をクリックします。

      説明

      複数のMaxComputeデータソースがDataStudioに関連付けられている場合は、使用するMaxComputeデータソースを選択する必要があります。

   4. テーブルの構成タブの上部ツールバーで、DDL. をクリックします。

   5. では、DDLダイアログボックスで、次のテーブル作成ステートメントを入力し、テーブルスキーマの生成. をクリックします。

      create table elastic2mc_bankdata 
      (
      age             string,
      job             string,
      marital         string,
      education       string,
      default         string,
      housing         string,
      loan            string,
      contact         string,
      month           string,
      day of week     string
      );

   6. 本番環境に送信.をクリックします。

3. データを同期します。

   1. データ分析ページに移動します。 指定したワークフローを右クリックし、[新規] > [データ統合] > [オフライン同期] を選択します。

   2. では、ノードの作成ダイアログボックスに名前を入力します。名前フィールドをクリックし、確認. をクリックします。

   3. 上部のナビゲーションバーで、アイコンが表示されます。

   4. スクリプトモードで、アイコンが表示されます。

   5. でテンプレートのインポートダイアログボックスソースタイプ,データソース,ターゲットタイプとデータソースをクリックし、確認.をクリックします。

   6. スクリプトを設定します。

      この例では、次のコードを使用しています。 コードの説明の詳細については、「Elasticsearch Reader」をご参照ください。

      {
       "type": "job",
       "steps": [
       {
       "stepType": "elasticsearch",
       "parameter": {
       "retryCount": 3,
       "column": [
       "age",
       "job",
       "marital",
       "education",
       "default",
       "housing",
       "loan",
       "contact",
       "month",
       "day_of_week",
       "duration",
       "campaign",
       "pdays",
       "previous",
       "poutcome",
       "emp_var_rate",
       "cons_price_idx",
       "cons_conf_idx",
       "euribor3m",
       "nr_employed",
       "y"
       ],
       "scroll": "1m",
       "index": "es_index",
       "pageSize": 1,
       "sort": {
       "age": "asc"
      },
       "type": "elasticsearch",
       "connTimeOut": 1000,
       "retrySleepTime": 1000,
       "endpoint": "http://es-cn-xxxx.xxxx.xxxx.xxxx.com:9200",
       "password": "xxxx",
       "search": {
       "match_all": {}
       },
       "readTimeOut": 5000,
       "username": "xxxx"
       },
       "name": "Reader",
       "category": "reader"
       },
       {
       "stepType": "odps",
       "parameter": {
       "partition": "",
       "truncate": true,
       "compress": false,
       "datasource": "odps_source",// The name of the MaxCompute data source.
       "column": [
       "age",
       "job",
       "marital",
       "education",
       "default",
       "housing",
       "loan",
       "contact",
       "month",
       "day_of_week",
       "duration",
       "campaign",
       "pdays",
       "previous",
       "poutcome",
       "emp_var_rate",
       "cons_price_idx",
       "cons_conf_idx",
       "euribor3m",
       "nr_employed",
       "y"
       ],
       "emptyAsNull": false,
       "table": "elastic2mc_bankdata"
       },
       "name": "Writer",
       "category": "writer"
       }
       ],
       "version": "2.0",
       "order": {
       "hops": [
       {
       "from": "Reader",
       "to": "Writer"
       }
       ]
       },
       "setting": {
       "errorLimit": {
       "record": "0"
       },
       "speed": {
       "throttle": false,
       "concurrent": 1,
       "dmu": 1
       }
       }
      }

      説明

      作成したElasticsearchクラスターの [基本情報] ページで、クラスターのパブリックエンドポイントとポート番号を表示できます。

   7. をクリックし、コードを実行するノードの構成タブの上部ツールバーにあるアイコン。

   8. の実行結果を表示します。ランタイムログタブをクリックします。

4. 結果を表示します。

   1. ワークフローを右クリックし、新しい > MaxCompute > ODPS SQL.

   2. でノードを作成するダイアログボックスは、ノード名をクリックし、送信.をクリックします。

   3. ODPS SQLノードの [設定] タブで、次のステートメントを入力します。

      SELECT * FROM elastic2mc_bankdata;

   4. アイコンをクリックしてコードを実行します。

   5. 操作ログで結果を表示します。