DataWorks:Elasticsearch data source

Offline synchronization task (single table)

• See Configure a task in the codeless UI or Configure a task in the code editor.

• For a full parameter reference and a script demo, see Appendix 1: Script demo and parameter descriptions.

Real-time write task (single table)

See Configure a real-time synchronization task in DataStudio (Legacy).

Full-database real-time synchronization

See Configure a real-time full-database synchronization task.

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 Script mode configuration. 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 allowed.
        },
        "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 context duration.
                "pageSize":5000,
                "connTimeOut":600000,
                "readTimeOut":600000,
                "retryCount":30,
                "retrySleepTime":"10000",
                "search":{
                            "range":{
                                "gmt_modified":{
                                    "gte":0
                                }
                            }
                        },  // Query parameter. The content is identical to the query body of the Elasticsearch _search API.
                "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 false, the mbps parameter is ignored and throttling is disabled. If throttle is true, throttling is enabled.
            "concurrent":1, // The job concurrency.
            "mbps":"12" // The maximum rate in Mbps (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 maps to _id in Elasticsearch and can be considered a unique primary key. When writing data, documents with the same id overwrite each other and are 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: Pulls an entire Elasticsearch document into a single field.

• Example configuration:

  
  ## reader: Original 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 configuration
  "parameter": {
    "column": [
        "content"
    ],
    "full":true
  }
  
  ## writer result: Writes one row and one column to the destination.
  {"_index":"mutiltest_1","_type":"_doc","_id":"IXgdO4MB4GR_1DmrjTXP","_source":{"feature1":"value1","feature2":"value2","feature3":"value3"},"sort":["IXgdO4MB4GR_1DmrjTXP"]}

Scenario 2: Synchronize nested or object field properties

• Background: Uses a path to access properties of an object or nested field.

• Syntax:

  • property

  • property.sub-property

  • property[0].sub-property

• Code editor configuration:

  "multi":{
      "multi":true
  }

  Note

  This feature is not supported in the codeless UI.

• Example configuration:

  ## reader: Original 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 configuration
  "parameter": {
    "column": [
        "level1",
        "level1.level2",
        "level1.level2[0]",
        "level1.level2.level3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## writer result: 1 row, 4 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 an ancestor node in the path is an array, the result is null. For example, retrieving level1.level2.level3 returns null because level2 is an array. To access the property, you must specify an index, such as level1.level2[0].level3 or level1.level2[1].level3. The wildcard syntax level1.level2[*].level3 is not supported.

  • Keys that contain a period (.) are not supported. For example, if the data is {"level1.level2":{"level3":"testlevel3_1"}}, the result is null.

Scenario 3: Split an array property into multiple rows

• Background: For one-to-many relationships, you can unroll an array by splitting each element into a separate row.

• Syntax: property[*].sub-property

• Example effect: Source data such as { "splitKey": [1, 2, 3, 4, 5] } is transformed into five rows, with each row containing one element from the array.

• Code editor configuration:

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

  Note

  • In the codeless UI, you can set the split multi-row array column name parameter to achieve the same effect by automatically generating the corresponding script configuration.

  • The value of the property specified by key must be a list. Otherwise, an error occurs.

• Example configuration:

  ## reader: Original 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 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: Deduplicates array elements and merges them into a comma-separated string. This also works with sub-properties such as name1.name2. Deduplication is based on the toString() result of each element.

• Syntax: property[].

  Including [] in the column name triggers this deduplication and merge operation.

• Code editor configuration:

  "multi":{
      "multi":true
  }

  Note

  This feature is not supported in the codeless UI.

• Example configuration:

  ## reader: Original 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 configuration
  "parameter": {
    "column":[
          "feature1[]"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## writer result: 1 row, 1 column
  "value1,value2,value3"

Scenario 5: Selectively synchronize multiple properties

• Background: Returns the first non-null property value from a specified list. If all properties are null or absent, null is returned.

• Syntax: property1|property2|...

  Including the pipe (|) character in the column name triggers this multi-property selection.

• Code editor configuration:

  "multi":{    
      "multi":true
  }

  Note

  This feature is not supported in the codeless UI.

• Example configuration:

  ## reader: Original 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 configuration
  "parameter": {
    "column":[
          "feature1|feature2|feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## writer result: 1 row, 1 column
  "feature1"

Scenario 6: Merge and sync multiple properties

• Background: Concatenates multiple property values into a single comma-separated string.

• Syntax: property1,property2,...

  Including a comma (,) in the column name triggers this multi-property merge operation.

• Code editor configuration:

  "multi":{
      "multi":true
  }

  Note

  This feature is not supported in the codeless UI.

• Example configuration:

  ## reader: Original 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 configuration
  "parameter": {
    "column":[
          "feature1,feature2,feature3"
    ],
    "multi":{
          "multi":true
      }
  }
  
  ## writer result: 1 row, 1 column
  "feature1,[1,2,3],{\"child\":\"feature3\"}"