大数据开发治理平台 DataWorks：Elasticsearch数据源

背景信息

Elasticsearch在公共资源组上支持Elasticsearch 5.x版本，在独享数据集成资源组和新版资源组（通用型资源组）上支持Elasticsearch 5.x、6.x、7.x和8.x版本。

Relational DB（实例）-> Databases（数据库）-> Tables（表）-> Rows（一行数据）-> Columns（一行数据的一列）
Elasticsearch        -> Index              -> Types       -> Documents       -> Fields

Elasticsearch中可以有多个索引或数据库，每个索引可以包括多个类型或表，每个类型可以包括多个文档或行，每个文档可以包括多个字段或列。Elasticsearch Writer插件使用Elasticsearch的Rest API接口，批量把从Reader读入的数据写入Elasticsearch中。

支持的版本

DataWorks平台目前仅支持配置阿里云Elasticsearch 5.x、6.x、7.x和8.x版本数据源，不支持配置自建Elasticsearch数据源。

使用限制

离线读写

• Elasticsearch Reader会获取Server端shard信息用于数据同步，需要确保在任务同步中Server端的shards处于存活状态，否则会存在数据不一致风险。

• 如果您使用的是6.x及以上版本，仅支持使用独享数据集成资源组或使用新版资源组。

• 不支持同步scaled_float类型的字段。

• 不支持同步字段中带有关键字 $ref的索引。

支持的字段类型

工作原理

更多详情请参见Elasticsearch官方文档。

基本配置

{
 "order":{
  "hops":[
   {
    "from":"Reader",
    "to":"Writer"
   }
  ]
 },
 "setting":{
  "errorLimit":{
   "record":"0" //错误记录数。
  },
  "jvmOption":"",
  "speed":{
   "concurrent":3,//并发数
   "throttle":true,//
                     "mbps":"12",//限流，此处1mbps = 1MB/s。
  }
 },
 "steps":[
  {
   "category":"reader",
   "name":"Reader",
   "parameter":{
    "column":[ //读取列。
     "id",
     "name"
    ],
    "endpoint":"", //服务地址。
    "index":"",  //索引。
    "password":"",  //密码。
    "scroll":"",  //scroll标志。
    "search":"",  //查询query参数，与Elasticsearch的query内容相同，使用_search api，重命名为search。
    "type":"default",
    "username":""  //用户名。
   },
   "stepType":"elasticsearch"
  },
  {
   "stepType": "elasticsearch",
            "parameter": {
                "column": [ //写入列
                    {
                        "name": "id",
                        "type": "integer"
                    }，
                    {
                        "name": "name",
                        "type": "text"
                    }
                ],
                "index": "test",   //写入索引
                 "indexType": "",   //写入索引类型，es7不填
                "actionType": "index",  //写入方式
                "cleanup": false,         //是否重建索引
                "datasource": "test",   //数据源名称
                "primaryKeyInfo": {     //主键取值方式
                    "fieldDelimiterOrigin": ",",
                    "column": [
                        "id"
                    ],
                    "type": "specific",
                    "fieldDelimiter": ","
                },
                "dynamic": false,  //动态映射
                "batchSize": 1024   //批量写文档数
            },
            "name": "Writer",
            "category": "writer"
  }
 ],
 "type":"job",
 "version":"2.0" //版本号。
}

高级功能

• 支持全量拉取

  支持将Elasticsearch中一个文档的所有内容拉取为一个字段。配置详情请参见场景一：全量拉取。

• 支持提取半结构化到结构化数据

  分类	描述	相关文档
  产生背景	Elasticsearch中的数据特征为字段不固定，且有中文名、数据使用深层嵌套的形式。为更好地方便下游业务对数据的计算和存储需求，特推出从半结构化到结构化的转换解决方案。	—
  实现原理	将Elasticsearch获取到的JSON数据，利用JSON工具的路径获取特性，将嵌套数据扁平化为一维结构的数据。然后将数据映射至结构化数据表中，拆分Elasticsearch复合结构数据至多个结构化数据表。	—
  解决方案	JSON有嵌套的情况，通过path路径来解决。 属性 属性.子属性 属性[0].子属性	场景二：嵌套或对象字段属性同步
  	附属信息有一对多的情况，需要进行拆表拆行处理，进行遍历。 属性[*].子属性	场景三：数组属性拆分为多行
  	数组归并，一个字符串数组内容，归并为一个属性，并进行去重。 属性[]	场景四：数组属性去重归并
  	多属性合一，将多个属性合并为一个属性。 属性1,属性2	场景五：多属性合一同步
  	多属性选择处理。 属性1|属性2	场景六：多属性选择同步

数据同步任务开发

Elasticsearch数据同步任务的配置入口和通用配置流程指导可参见下文的配置指导，详细的配置参数解释可在配置界面查看对应参数的文案提示。

创建数据源

在进行数据同步任务开发时，您需要在DataWorks上创建一个对应的数据源，操作流程请参见创建并管理数据源。

单表离线同步任务配置指导

• 操作流程请参见通过向导模式配置离线同步任务、通过脚本模式配置离线同步任务。

• 脚本模式配置的全量参数和脚本Demo请参见下文的附录：脚本Demo与参数说明。

单表实时写同步任务配置指导

操作流程请参见DataStudio侧实时同步任务配置。

整库离线写、单表/整库全增量实时写同步任务配置指导

操作流程请参见数据集成侧同步任务配置。

附录：脚本Demo与参数说明

附录：离线任务脚本配置方式

如果您配置离线任务时使用脚本模式的方式进行配置，您需要在任务脚本中按照脚本的统一格式要求编写脚本中的reader参数和writer参数，脚本模式的统一要求请参见通过脚本模式配置离线同步任务，以下为您介绍脚本模式下的数据源的Reader参数和Writer参数的指导详情。

Reader脚本Demo

{
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    },
    "setting":{
        "errorLimit":{
            "record":"0" //错误记录数。
        },
        "jvmOption":"",
        "speed":{
            "concurrent":3,
            "throttle":false
        }
    },
    "steps":[
        {
            "category":"reader",
            "name":"Reader",
            "parameter":{
                "column":[ //读取列。
                    "id",
                    "name"
                ],
                "endpoint":"http://es-cn-xxx.elasticsearch.aliyuncs.com:9200", //服务地址。
                "index":"aliyun_es_xx",  //索引。
                "password":"*******",  //密码。
                "multiThread":true,
                "scroll":"5m",  //scroll标志。
                "pageSize":5000,
                "connTimeOut":600000,
                "readTimeOut":600000,
                "retryCount":30,
                "retrySleepTime":"10000",
                "search":{
                            "range":{
                                "gmt_modified":{
                                    "gte":0
                                }
                            }
                        },  //查询query参数，与Elasticsearch的query内容相同，使用_search api，重命名为search。
                "type":"doc",
                "username":"aliyun_di"  //用户名。
            },
            "stepType":"elasticsearch"
        },
        {
            "category":"writer",
            "name":"Writer",
            "parameter":{ },
            "stepType":"stream"
        }
    ],
    "type":"job",
    "version":"2.0" //版本号。
}

Reader脚本参数

Writer脚本Demo

{
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent":1, //作业并发数。
            "mbps":"12"//限流，此处1mbps = 1MB/s。
        }
    },
    "steps": [
        {
            "category": "reader",
            "name": "Reader",
            "parameter": {

            },
            "stepType": "stream"
        },
        {
            "category": "writer",
            "name": "Writer",
            "parameter": {
                "datasource":"xxx",
                "index": "test-1",
                "type": "default",
                "cleanup": true,
                "settings": {
                        "number_of_shards": 1,
                        "number_of_replicas": 0
                },
                "discovery": false,
                "primaryKeyInfo":{
                    "type":"pk",    
                     "fieldDelimiter":",",
                     "column":[]
                    },
                "batchSize": 1000,
                "dynamic":false,
                "esPartitionColumn":[
                    {
                        "name":"col1",  
                        "comment":"xx", 
                        "type":"STRING" 
                        }
                     ],
                "column": [
                    {
                        "name": "pk",
                        "type": "id"
                    },
                    {
                        "name": "col_ip",
                        "type": "ip"
                    },
                    {
                        "name": "col_array",
                        "type": "long",
                        "array": true,
                    },
                    {
                        "name": "col_double",
                        "type": "double"
                    },
                    {
                        "name": "col_long",
                        "type": "long"
                    },
                    {
                        "name": "col_integer",
                        "type": "integer"
                    {
                        "name": "col_keyword",
                        "type": "keyword"
                    },
                    {
                        "name": "col_text",
                        "type": "text",
                        "analyzer": "ik_max_word",
                        "other_params":
                            {
                                "doc_values": false
                            },
                    },
                    {
                        "name": "col_geo_point",
                        "type": "geo_point"
                    },
                    {
                        "name": "col_date",
                        "type": "date",
                        "format": "yyyy-MM-dd HH:mm:ss"
                    },
                    {
                        "name": "col_nested1",
                        "type": "nested"
                    },
                    {
                        "name": "col_nested2",
                        "type": "nested"
                    },
                    {
                        "name": "col_object1",
                        "type": "object"
                    },
                    {
                        "name": "col_object2",
                        "type": "object"
                    },
                    {
                        "name": "col_integer_array",
                        "type": "integer",
                        "array": true
                    },
                    {
                        "name": "col_geo_shape",
                        "type": "geo_shape",
                        "tree": "quadtree",
                        "precision": "10m"
                    }
                ]
            },
            "stepType": "elasticsearch"
        }
    ],
    "type": "job",
    "version": "2.0"
}

Writer脚本参数

- id  //type id对应Elasticsearch中的_id，可以理解为唯一主键。写入时，相同id的数据会被覆盖，且不会被索引。
- string
- text
- keyword
- long
- integer
- short
- byte
- double
- float
- date
- boolean
- binary
- integer_range
- float_range
- long_range
- double_range
- date_range
- geo_point
- geo_shape
- ip
- token_count
- array
- object
- nested

 {
   "name": "guid",
   "other_params":
    {
       "doc_values": false
      },
    "type": "text"
  }

场景一：全量拉取

• 背景说明：将Elasticsearch中文档查询的结果拉取为一个字段。

• 配置示例：

  
  ## 读端：Elasticsearch中的原始数据
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "IXgdO4MB4GR_1DmrjTXP",
          "_score": 1.0,
          "_source": {
              "feature1": "value1",
              "feature2": "value2",
              "feature3": "value3"
          }
      }]
  
  ##数据集成Elasticsearch Reader插件配置
  "parameter": {
    "column": [
        "content"
    ],
    "full":true
  }
  
  ##写端结果：同步至目标端1行1列
  {"_index":"mutiltest_1","_type":"_doc","_id":"IXgdO4MB4GR_1DmrjTXP","_source":{"feature1":"value1","feature2":"value2","feature3":"value3"},"sort":["IXgdO4MB4GR_1DmrjTXP"]}

场景二：嵌套或对象字段属性同步

• 背景说明：Object对象或nested嵌套字段的属性时，通过path路径来解决。

• 配置形式：

  • 属性

  • 属性.子属性

  • 属性[0].子属性

• 脚本配置：

  "multi":{
      "multi":true
  }

  说明

  向导模式暂不支持配置。

• 配置示例：

  ## 读端：Elasticsearch中的原始数据
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "7XAOOoMB4GR_1Dmrrust",
          "_score": 1.0,
          "_source": {
              "level1": {
                  "level2": [
                      {
                          "level3": "testlevel3_1"
                      },
                      {
                          "level3": "testlevel3_2"
                      }
                  ]
              }
          }
      }
  ]
  ##数据集成Elasticsearch reader插件配置
  "parameter": {
    "column": [
        "level1",
        "level1.level2",
        "level1.level2[0]",
        "level1.level2.level3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ##写端结果：1行数据4列
  column1(level1):            {"level2":[{"level3":"testlevel3_1"},{"level3":"testlevel3_2"}]}
  column2(level1.level2):     [{"level3":"testlevel3_1"},{"level3":"testlevel3_2"}]
  column3(level1.level2[0]):  {"level3":"testlevel3_1"}
  column4(level1.level2.level3):  null

  说明

  • 获取的节点上层有数组时结果为null，如上样例获取level1.level2.level3不会报错，同步结果为null，需要配置为level1.level2[0].level3或level1.level2[1].level3，当前不支持level1.level2[*].level3。

  • 不支持key出现"."的数据, 如"level1.level2":{"level3":"testlevel3_1"}, 此时该条数据获取结果为null。

场景三：数组属性拆分为多行

• 背景说明：附属信息有一对多的情况，需要将数组列拆成多行。

• 配置形式：属性[*].子属性

• 效果示意：源端数据{ "splitKey" :[1,2,3,4,5]}，拆完后写到目标端为5行：{"splitKey[0]":1,"splitKey[1]":2,"splitKey[2]":3,"splitKey[3]":4,"splitKey[4]":5}

• 脚本配置：

  "multi":{   
         "multi":true,    
          "key": "headers"
  }

  说明

  • 向导模式下配置拆多行数组列名，会自动生成脚本配置，具有相同效果。

  • value必须为List，否则会报错。

• 配置示例：

  ## 读端：Elasticsearch中的原始数据
  [
      {
          "_index": "lmtestjson",
          "_type": "_doc",
          "_id": "nhxmIYMBKDL4VkVLyXRN",
          "_score": 1.0,
          "_source": {
              "headers": [
                  {
                      "remoteip": "192.0.2.1"
                  },
                  {
                      "remoteip": "192.0.2.2"
                  }
              ]
          }
      },
      {
          "_index": "lmtestjson",
          "_type": "_doc",
          "_id": "wRxsIYMBKDL4VkVLcXqf",
          "_score": 1.0,
          "_source": {
              "headers": [
                  {
                      "remoteip": "192.0.2.3"
                  },
                  {
                      "remoteip": "192.0.2.4"
                  }
              ]
          }
      }
  ]
  ##数据集成Elasticsearch reader插件配置
  {
     "column":[
        "headers[*].remoteip"
    ]
    "multi":{
        "multi":true,
        "key": "headers"
    }
  }
  
  ##写端结果：4行
  192.0.2.1
  192.0.2.2
  192.0.2.3
  192.0.2.4

场景四：数组属性去重归并

• 背景说明：数组去重归并，将一个数组属性去重归并后写入为字符串属性，数组属性可以为子属性如name1.name2，去重采用tostring结果作为标准。

• 配置形式：属性[]。

  column里面带有 [] 关键字就会认为对该属性做去重归并。

• 脚本配置：

  "multi":{
      "multi":true
  }

  说明

  向导模式暂不支持配置。

• 配置示例：

  ## 读端：Elasticsearch中的原始数据
  "hits": [
  {
      "_index": "mutiltest_1",
      "_type": "_doc",
      "_id": "4nbUOoMB4GR_1Dmryj8O",
      "_score": 1.0,
      "_source": {
          "feature1": [
              "value1",
              "value1",
              "value2",
              "value2",
              "value3"
          ]
      }
  }
  ]
  ##数据集成Elasticsearch reader插件配置
  "parameter": {
    "column":[
          "feature1[]"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ##写端结果：1行1列数据
  "value1,value2,value3"

场景五：多属性合一同步

• 背景说明：多属性选择处理，返回第一个有值的属性，都不存在时将写入null。

• 配置形式：属性1|属性2|...

  column里面带有 "|"关键字就会对该项做多属性选择。

• 脚本配置：

  "multi":{    
      "multi":true
  }

  说明

  向导模式暂不支持该配置。

• 配置示例：

  ##读端：Elasticsearch中的原始数据
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "v3ShOoMB4GR_1DmrZN22",
          "_score": 1.0,
          "_source": {
              "feature1": "feature1",
              "feature2": [
                  1,
                  2,
                  3
              ],
              "feature3": {
                  "child": "feature3"
              }
          }
      }]
  
  ##数据集成Elasticsearch reade插件配置
  "parameter": {
    "column":[
          "feature1|feature2|feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ##写端结果：1行1列数据
  "feature1"

场景六：多属性选择同步

• 背景说明：多属性选择处理 ,返回第一个有值的属性，都不存在时写入null。

• 配置形式：属性1|属性2|...

  column里面带有 "|"关键字就会对该项做多属性选择

• 脚本配置：

  "multi":{
      "multi":true
  }

  说明

  向导模式暂不支持该配置。

• 配置示例：

  ##读端：Elasticsearch中的原始数据
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "v3ShOoMB4GR_1DmrZN22",
          "_score": 1.0,
          "_source": {
              "feature1": "feature1",
              "feature2": [
                  1,
                  2,
                  3
              ],
              "feature3": {
                  "child": "feature3"
              }
          }
      }]
  ##数据集成Elasticsearch reader插件配置
  "parameter": {
    "column":[
          "feature1,feature2,feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ##写端结果：1行1列数据
  "feature1,[1,2,3],{"child":"feature3"}"

附录：ElasticSearch写入的格式期望是数组类型

支持以下两种方式将源端数据按照数组类型写入ElasticSearch。

• 按JSON格式解析源端数据

  例如：源端数据为"[1,2,3,4,5]"，配置json_array=true对其进行解析，同步将以数组格式写入ElasticSearch。

  "parameter" : {
    {
      "name":"docs_1",
      "type":"keyword",
      "json_array":true
    }
  }

• 按分隔符解析源端数据

  例如：源端数据为"1,2,3,4,5", 配置分隔符splitter=","对其进行解析，同步将以数组格式写入ElasticSearch。

  说明

  一个任务仅支持配置一种分隔符，splitter全局唯一，不支持多array字段配置为不同的分隔符。例如源端字段列col1="1,2,3,4,5" , col2="6-7-8-9-10"， splitter无法针对每列单独配置使用。

  "parameter" : {
        "column": [
          {
            "name": "docs_2",
            "array": true,
            "type": "long"
          }
        ],
        "splitter":","//注意：splitter配置与column配置同级。
  }

相关文档

数据集成支持其他更多数据源接入，更多信息，请参见支持的数据源及同步方案。