DataWorks:Elasticsearch data source

Configure an offline synchronization task for a single table

• For more information about the procedure, see Configure a batch synchronization task using the codeless UI and Configuration in the code editor.

• For all parameters and a script demo for the code editor, see Appendix 1: Script demos and parameter descriptions.

Configure a real-time write task for a single table

For more information about the procedure, see Configure a real-time synchronization task in DataStudio.

Configure an offline write task for an entire database or a full/incremental real-time write task for a single table or an entire database

For more information about the procedure, see Configure a real-time synchronization task for an entire database.

Configure a batch synchronization task by using the code editor

If you want to configure a batch synchronization task by using the code editor, you must configure the related parameters in the script based on the unified script format requirements. For more information, see Configuration in the code editor. The following information describes the parameters that you must configure for data sources when you configure a batch synchronization task by using the code editor.

Reader script demo

{
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    },
    "setting":{
        "errorLimit":{
            "record":"0" // The number of error records.
        },
        "jvmOption":"",
        "speed":{
            "concurrent":3,
            "throttle":false
        }
    },
    "steps":[
        {
            "category":"reader",
            "name":"Reader",
            "parameter":{
                "column":[ // The columns to read.
                    "id",
                    "name"
                ],
                "endpoint":"http://es-cn-xxx.elasticsearch.aliyuncs.com:9200", // The endpoint.
                "index":"aliyun_es_xx",  // The index.
                "password":"*******",  // The password.
                "multiThread":true,
                "scroll":"5m",  // The scroll flag.
                "pageSize":5000,
                "connTimeOut":600000,
                "readTimeOut":600000,
                "retryCount":30,
                "retrySleepTime":"10000",
                "search":{
                            "range":{
                                "gmt_modified":{
                                    "gte":0
                                }
                            }
                        },  // The query parameter. This is the same as the query content of Elasticsearch. The _search API is used and renamed to search.
                "type":"doc",
                "username":"aliyun_di"  // The username.
            },
            "stepType":"elasticsearch"
        },
        {
            "category":"writer",
            "name":"Writer",
            "parameter":{ },
            "stepType":"stream"
        }
    ],
    "type":"job",
    "version":"2.0" // The version number.
}

Reader script parameters

Writer script demo

{
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle":true,// If throttle is set to false, the mbps parameter does not take effect, which means no throttling. If throttle is set to true, throttling is enabled.
            "concurrent":1, // The number of concurrent jobs.
            "mbps":"12"// Throttling. 1 Mbps = 1 MB/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 script parameters

- id  // The id type corresponds to _id in Elasticsearch, which can be understood as a unique primary key. When writing data, data with the same ID is overwritten and not indexed.
- 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"
  }

Scenario 1: Full data pull

• Background: You can pull the query results of a document in Elasticsearch as a single field.

• Example:

  
  ## Reader: Raw data in Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "IXgdO4MB4GR_1DmrjTXP",
          "_score": 1.0,
          "_source": {
              "feature1": "value1",
              "feature2": "value2",
              "feature3": "value3"
          }
      }]
  
  ## Data Integration Elasticsearch Reader plug-in configuration
  "parameter": {
    "column": [
        "content"
    ],
    "full":true
  }
  
  ## Writer result: One row and one column are synchronized to the destination.
  {"_index":"mutiltest_1","_type":"_doc","_id":"IXgdO4MB4GR_1DmrjTXP","_source":{"feature1":"value1","feature2":"value2","feature3":"value3"},"sort":["IXgdO4MB4GR_1DmrjTXP"]}

Scenario 2: Synchronize properties of nested or object fields

• Background: When you synchronize properties of Object or nested fields, you can use a path to resolve them.

• Configuration:

  • property

  • property.sub-property

  • property[0].sub-property

• Script configuration:

  "multi":{
      "multi":true
  }

  Note

  Configuration is not currently available in the codeless UI.

• Example:

  ## Reader: Raw data in Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "7XAOOoMB4GR_1Dmrrust",
          "_score": 1.0,
          "_source": {
              "level1": {
                  "level2": [
                      {
                          "level3": "testlevel3_1"
                      },
                      {
                          "level3": "testlevel3_2"
                      }
                  ]
              }
          }
      }
  ]
  ## Data Integration Elasticsearch reader plug-in configuration
  "parameter": {
    "column": [
        "level1",
        "level1.level2",
        "level1.level2[0]",
        "level1.level2.level3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Writer result: One row of data with four columns
  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

  Note

  • If a parent node of the retrieved node contains an array, the result is null. For example, retrieving `level1.level2.level3` does not report an error, but the synchronization result is null. You must configure the path as `level1.level2[0].level3` or `level1.level2[1].level3`. The `level1.level2[*].level3` format is not currently supported.

  • Data that contains a period (.) in the key is not supported, such as `{"level1.level2":{"level3":"testlevel3_1"}}`. In this case, the retrieval result for this data entry is null.

Scenario 3: Split array properties into multiple rows

• Background: If a one-to-many relationship exists, you can split the array column into multiple rows.

• Configuration: property[*].sub-property

• Effect: Source data such as `{ "splitKey" :[1,2,3,4,5]}` is split and written to the destination as five separate rows.

• Script configuration:

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

  Note

  • In the codeless UI, configuring Split Multi-row Array Column Name automatically generates an equivalent script configuration.

  • The value must be a list. Otherwise, an error occurs.

• Example:

  ## Reader: Raw data in 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"
                  }
              ]
          }
      }
  ]
  ## Data Integration Elasticsearch reader plug-in configuration
  {
     "column":[
        "headers[*].remoteip"
    ]
    "multi":{
        "multi":true,
        "key": "headers"
    }
  }
  
  ## Writer result: 4 rows
  192.0.2.1
  192.0.2.2
  192.0.2.3
  192.0.2.4

Scenario 4: Deduplicate and merge array properties

• Background: You can deduplicate and merge an array property, and then write the result as a string property. The array property can be a sub-property, such as `name1.name2`. The `toString` result is used as the standard for deduplication.

• Configuration: `property[]`.

  If the column name contains `[]`, deduplication and merging are performed on this property.

• Script configuration:

  "multi":{
      "multi":true
  }

  Note

  This parameter cannot be configured in the codeless UI.

• Example:

  ## Reader: Raw data in Elasticsearch
  "hits": [
  {
      "_index": "mutiltest_1",
      "_type": "_doc",
      "_id": "4nbUOoMB4GR_1Dmryj8O",
      "_score": 1.0,
      "_source": {
          "feature1": [
              "value1",
              "value1",
              "value2",
              "value2",
              "value3"
          ]
      }
  }
  ]
  ## Data Integration Elasticsearch reader plug-in configuration
  "parameter": {
    "column":[
          "feature1[]"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Writer result: One row and one column of data
  "value1,value2,value3"

Scenario 5: Merge and synchronize multiple properties

• Background: You can selectively process multiple properties. The first property that has a value is returned. If none of the specified properties exist, null is written.

• Configuration: `property1|property2|...`

  If the column name contains the `|` separator, multiple properties are selected for this item.

• Script configuration:

  "multi":{    
      "multi":true
  }

  Note

  This parameter cannot be configured in the codeless UI.

• Example:

  ## Reader: Raw data in Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "v3ShOoMB4GR_1DmrZN22",
          "_score": 1.0,
          "_source": {
              "feature1": "feature1",
              "feature2": [
                  1,
                  2,
                  3
              ],
              "feature3": {
                  "child": "feature3"
              }
          }
      }]
  
  ## Data Integration Elasticsearch reader plug-in configuration
  "parameter": {
    "column":[
          "feature1|feature2|feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Writer result: One row and one column of data
  "feature1"

Scenario 6: Selectively synchronize multiple properties

• Background: You can selectively process multiple properties. The first property that has a value is returned. If none of the specified properties exist, null is written.

• Configuration: `property1|property2|...`

  If the column name contains the `|` separator, multiple properties are selected for this item.

• Script configuration:

  "multi":{
      "multi":true
  }

  Note

  This parameter cannot be configured in the codeless UI.

• Example:

  ## Reader: Raw data in Elasticsearch
  "hits": [
      {
          "_index": "mutiltest_1",
          "_type": "_doc",
          "_id": "v3ShOoMB4GR_1DmrZN22",
          "_score": 1.0,
          "_source": {
              "feature1": "feature1",
              "feature2": [
                  1,
                  2,
                  3
              ],
              "feature3": {
                  "child": "feature3"
              }
          }
      }]
  ## Data Integration Elasticsearch reader plug-in configuration
  "parameter": {
    "column":[
          "feature1,feature2,feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## Writer result: One row and one column of data
  "feature1,[1,2,3],{"child":"feature3"}"