：Elasticsearch Reader

使用限制

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

支持的Elasticsearch版本

支持的字段类型

工作原理

Elasticsearch在公共资源组上支持Elasticsearch5.x版本，在独享数据集成资源组上支持Elasticsearch5.x、6.x和7.x版本。独享数据集成资源组的详情请参见新增和使用独享数据集成资源组。

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

基本配置

{
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    },
    "setting":{
        "errorLimit":{
            "record":"0" //错误记录数。
        },
        "jvmOption":"",
        "speed":{
            "concurrent":3,//并发数
            "throttle":true,//
                     "mbps":"12",//限流
        }
    },
    "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"
        },
        {
            "category":"writer",
            "name":"Writer",
            "parameter":{ },
            "stepType":"stream"
        }
    ],
    "type":"job",
    "version":"2.0" //版本号。
}

高级功能

• 支持全量拉取

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

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

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

参数说明

向导开发介绍

打开新建的数据同步节点，即可进行同步任务的配置，详情请参见通过向导模式配置离线同步任务。

脚本开发介绍

{
    "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" //版本号。
}

配置数据集成资源组

1. 单击数据同步任务编辑页面右侧的数据集成资源组配置。
2. 根据提示选择对应的独享数据集成资源组。
   

   说明

   • （推荐）数据集成资源组配置页面默认支持选择独享数据集成资源组，为确保数据同步的稳定性和性能要求，推荐使用独享数据集成资源组。
   • 如果您需要选择公共资源组，请在页面右下方单击更多选项，在弹出的警告对话框单击确认，在数据集成资源组配置子页面进行选择。关于自定义数据集成资源组和公共资源组，详情请参见公共资源组。

场景一：全量拉取

• 背景说明：将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"}"