通过数据集成导入数据至Elasticsearch

新建离线同步节点

1. 登录DataWorks控制台，单击相应工作空间后的进入数据集成。
2. 在数据集成 > 首页页面，单击新建同步任务，跳转至数据开发页面。
3. 在新建节点对话框中，输入节点名称并选择目标文件夹，单击提交。
   说明

   • 节点名称的长度不能超过128个字符。
   • 此处的目标文件夹为创建的业务流程，具体操作请参见新建业务流程。

配置离线同步节点

1. 成功创建离线同步节点后，单击工具栏中的转换脚本。
   
2. 单击提示对话框中的确认，即可进入脚本模式进行开发。
3. 参照下文的脚本模板，根据自身需求进行配置。

   {
   "configuration": {
   "setting": {
     "speed": {
       "concurrent": "1", //作业并发数。
       "mbps": "1" //作业速率上限。
     }
   },
   "reader": {
     "parameter": {
       "connection": [
         {
           "table": [
             "`es_table`" //源端表名。
           ],
           "datasource": "px_mysql_OK" //数据源名，建议和添加的数据源名保持一致。
         }
       ],
       "column": [ //源端表的列名。
         "col_ip",
         "col_double",
         "col_long",
         "col_integer",
         "col_keyword",
         "col_text",
         "col_geo_point",
         "col_date"
       ],
       "where": "", //过滤条件。
     },
     "plugin": "mysql"
   },
   "writer": {
     "parameter": {
       "cleanup": true, //是否在每次导入数据到Elasticsearch时清空原有数据，全量导入或重建索引时，需要设置为true，同步增量时必须为false。
       "accessKey": "nimda", //如果使用了X-PACK插件，需要填写password；如果未使用，则填空字符串即可。阿里云Elasticsearch使用了X-PACK插件，需要填写password。
       "index": "datax_test", // Elasticsearch的索引名称，如果之前没有，插件会自动创建。
       "alias": "test-1-alias", //数据导入完成后写入别名。
       "settings": {
         "index": {
           "number_of_replicas": 0,
           "number_of_shards": 1
         }
       },
       "batchSize": 1000, //每次批量数据的条数。
       "accessId": "default", //如果使用了X-PACK插件，需要填写username；如果未使用，则填空字符串即可。阿里云Elasticsearch使用了X-PACK插件，需要填写username。
       "endpoint": "http://xxx.xxxx.xxx:xxxx", //Elasticsearch的连接地址，可以在控制台查看。
       "splitter": ",", //如果插入数据是array，则使用指定分隔符。
       "indexType": "default", //Elasticsearch中相应索引下的类型名称。
       "aliasMode": "append", //数据导入完成后增加别名的模式，append（增加模式），exclusive（只留这一个）。
       "column": [ //Elasticsearch中的列名，顺序和Reader中的Column顺序一致。
         {
           "name": "col_ip",//对应于TableStore中的属性列：name。
           "type": "ip"//文本类型，采用默认分词。
         },
         {
           "name": "col_double",
           "type": "string"
         },
         {
           "name": "col_long",
           "type": "long"
         },
         {
           "name": "col_integer",
           "type": "integer"
         },
         {
           "name": "col_keyword",
           "type": "keyword"
         },
         {
           "name": "col_text",
           "type": "text"
         },
         {
           "name": "col_geo_point",
           "type": "geo_point"
         },
         {
           "name": "col_date",
           "type": "date"
         }
       ],
       "discovery": false//是否自动发现，设置为true。
     },
     "plugin": "elasticsearch"//Writer插件的名称：ElasticsearchWriter，无需修改。
   }
   },
   "type": "job",
   "version": "1.0"
   }

4. 单击后，再单击。
   说明

   • Elasticsearch仅支持以脚本模式导入数据。
   • 保存同步任务后，直接单击，任务会立刻运行。

     您也可以单击，提交同步任务至调度系统中，调度系统会按照配置属性在从第2天开始自动定时执行。

后续步骤

如果您需要使用其它类型的数据源配置同步任务，请参见配置Reader和配置Writer模块的文档。