DataWorks:Elasticsearch

Configure an offline sync task (single table)

• For a step-by-step walkthrough, see Configure a task in the codeless UI or Configure a task in the code editor.

• For the full parameter reference and script examples, see Appendix 1: Script demos and parameter descriptions.

Configure a real-time write task (single table)

See Configure a real-time synchronization task in DataStudio.

Configure a full-database offline or real-time sync task

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

Reader script 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",
        "pageSize": 5000,
        "connTimeOut": 600000,
        "readTimeOut": 600000,
        "retryCount": 30,
        "retrySleepTime": "10000",
        "search": {
          "range": {
            "gmt_modified": {
              "gte": 0
            }
          }
        },
        "type": "doc",
        "username": "aliyun_di"
      },
      "stepType": "elasticsearch"
    },
    {
      "category": "writer",
      "name": "Writer",
      "parameter": {},
      "stepType": "stream"
    }
  ],
  "type": "job",
  "version": "2.0"
}

Reader script parameters

Writer script demo

An Elasticsearch instance inside a Virtual Private Cloud (VPC) cannot be reached from a public resource group due to network isolation. Use a serverless resource group or an exclusive resource group for Data Integration to access instances in a VPC.

{
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  },
  "setting": {
    "errorLimit": {
      "record": "0"
    },
    "speed": {
      "throttle": true,
      "concurrent": 1,
      "mbps": "12"
    }
  },
  "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          // Maps to _id in Elasticsearch. Documents with the same ID are overwritten, not re-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

Parse as JSON

If the source data is already a JSON array string such as "[1,2,3,4,5]", set "json_array": true to parse it:

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

Parse with a separator

If the source data is a delimited string such as "1,2,3,4,5", set "array": true on the column and configure splitter at the parameter level:

{
  "parameter": {
    "column": [
      {
        "name": "docs_2",
        "array": true,
        "type": "long"
      }
    ],
    "splitter": ","    // Must be at the same level as "column", not inside it.
  }
}

splitter is a global parameter — a task supports only one separator. If multiple array fields use different delimiters (for example, col1="1,2,3" and col2="6-7-8"), you cannot configure separate splitters for each field.

Scenario 1: Full data pull

Pull an entire Elasticsearch document as a single JSON string field. Set "full": true in the Reader parameter.

Input — raw Elasticsearch document:

"hits": [
  {
    "_index": "mutiltest_1",
    "_type": "_doc",
    "_id": "IXgdO4MB4GR_1DmrjTXP",
    "_score": 1.0,
    "_source": {
      "feature1": "value1",
      "feature2": "value2",
      "feature3": "value3"
    }
  }
]

Reader configuration:

"parameter": {
  "column": ["content"],
  "full": true
}

Output — one row, one column:

{"_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

Use dot notation and array index notation to extract specific properties from object or nested fields:

• property — top-level property

• property.sub-property — nested sub-property

• property[0].sub-property — element at a specific array index

Configuration is not available in the codeless UI for this scenario. Use the code editor.

Reader configuration:

"multi": {
  "multi": true
}

Input — raw Elasticsearch document:

"hits": [
  {
    "_index": "mutiltest_1",
    "_type": "_doc",
    "_id": "7XAOOoMB4GR_1Dmrrust",
    "_score": 1.0,
    "_source": {
      "level1": {
        "level2": [
          { "level3": "testlevel3_1" },
          { "level3": "testlevel3_2" }
        ]
      }
    }
  }
]

Column configuration:

"column": [
  "level1",
  "level1.level2",
  "level1.level2[0]",
  "level1.level2.level3"
]

Output — one row, 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

If a parent node in the path is an array, the result is null. For example, level1.level2.level3 returns null because level2 is an array. Use level1.level2[0].level3 or level1.level2[1].level3 instead. The level1.level2[*].level3 wildcard is not currently supported.

Keys that contain a period (.) are not supported. For example, {"level1.level2": {"level3": "value"}} returns null for this path.

Scenario 3: Split array properties into multiple rows

Expand a one-to-many array into separate rows using the [*] wildcard in the column path.

Configuration: property[*].sub-property

In the codeless UI, set Split multi-row array column name to generate the equivalent script configuration.

Reader configuration:

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

The value of key must be an array field. If it is not, an error occurs.

Input — two Elasticsearch documents:

[
  {
    "_source": {
      "headers": [
        { "remoteip": "192.0.2.1" },
        { "remoteip": "192.0.2.2" }
      ]
    }
  },
  {
    "_source": {
      "headers": [
        { "remoteip": "192.0.2.3" },
        { "remoteip": "192.0.2.4" }
      ]
    }
  }
]

Column configuration:

"column": ["headers[*].remoteip"]

Output — four rows:

192.0.2.1
192.0.2.2
192.0.2.3
192.0.2.4

Scenario 4: Deduplicate and merge array properties

Deduplicate and merge an array field into a single comma-separated string. Include [] in the column name to activate deduplication. Deduplication uses the toString result as the comparison key.

Configuration: property[]

Configuration is not available in the codeless UI. Use the code editor.

Reader configuration:

"multi": {
  "multi": true
}

Input:

"_source": {
  "feature1": ["value1", "value1", "value2", "value2", "value3"]
}

Column configuration:

"column": ["feature1[]"]

Output — one row, one column:

"value1,value2,value3"

Scenario 5: Merge and synchronize multiple properties

Merge multiple properties into a single column. All named properties are concatenated into one value.

Configuration: property1,property2,...

Configuration is not available in the codeless UI. Use the code editor.

Reader configuration:

"multi": {
  "multi": true
}

Input:

"_source": {
  "feature1": "feature1",
  "feature2": [1, 2, 3],
  "feature3": { "child": "feature3" }
}

Column configuration:

"column": ["feature1,feature2,feature3"]

Output — one row, one column:

"feature1,[1,2,3],{"child":"feature3"}"

Scenario 6: Selectively synchronize multiple properties

Return the first non-null value from a list of properties. If none of the specified properties have a value, null is written.

Configuration: property1|property2|...

Configuration is not available in the codeless UI. Use the code editor.

Reader configuration:

"multi": {
  "multi": true
}

Input:

"_source": {
  "feature1": "feature1",
  "feature2": [1, 2, 3],
  "feature3": { "child": "feature3" }
}

Column configuration:

"column": ["feature1|feature2|feature3"]

Output — one row, one column (first non-null value):

"feature1"