DataWorks:Migrate data from Elasticsearch to MaxCompute

Procedure

1. Create a source table in Elasticsearch. For more information, see Use DataWorks to synchronize data from MaxCompute to Alibaba Cloud Elasticsearch.

2. Create a destination table in MaxCompute.

   1. Log on to the DataWorks console. In the left-side navigation pane, choose Data Modeling and Development > DataStudio. On the page that appears, select the desired workspace from the drop-down list and click Go to DataStudio.

   2. In the left-side navigation pane, click Workspaces.

   3. In the Create Table dialog box, configure Name and click Create.

      Note

      If multiple MaxCompute compute engine instances are associated with the current workspace, you must select one from the Engine Instance drop-down list.

   4. On the table editing page, click DDL Statement.

   5. In the DDL dialog box, enter the following table creation statement and click Generate Table Schema.

      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. Click Submit to Production Environment.

3. Synchronize data.

   1. Go to the data analytics page. Right-click the specified workflow and choose new > data integration > offline synchronization.

   2. In the Create Node dialog box, enter a name in the Name field and click Confirm.

   3. In the top navigation bar, choose icon.

   4. In script mode, click icon.

   5. In import Template dialog box SOURCE type, data source, target type and data source, and click confirm.

   6. Configure the script.

      The following code is used in this example. For more information about the code description, see 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
       }
       }
      }

      Note

      On the Basic Information page of the created Alibaba Cloud Elasticsearch cluster, you can view the public endpoint and port number of the cluster.

   7. Click the icon to run the code.

   8. View the execution result on the Runtime Logs tab.

4. View the result.

   1. Right-click the workflow and choose new > MaxCompute > ODPS SQL.

   2. In create a node dialog box, enter node name, and click submit.

   3. On the configuration tab of the ODPS SQL node, enter the following statement:

      SELECT * FROM elastic2mc_bankdata;

   4. Click icon to run the code.

   5. You can operation Log view the results.