Elasticsearch数据源

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

• 操作流程请参见向导模式配置、脚本模式配置。

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

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

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

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

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

离线任务脚本配置方式

如果您配置离线任务时使用脚本模式的方式进行配置，您需要按照统一的脚本格式要求，在任务脚本中编写相应的参数，详情请参见脚本模式配置，以下为您介绍脚本模式下数据源的参数配置详情。

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"}"