How to create an Elasticsearch synchronization task-阿里云帮助中心

Before you begin

Create an Elasticsearch index
DataHub supports synchronizing data to an index in an Elasticsearch instance. This feature is compatible with Elasticsearch versions 5, 6, and 7.
Currently, DataHub can only synchronize data from TUPLE-type topics to Elasticsearch. Before you start the synchronization task, ensure that you have created an index in Elasticsearch or that automatic index creation is enabled. Otherwise, the synchronization task will fail.
Prepare and authorize an account for the synchronization task
To create an Elasticsearch synchronization task, provide your Elasticsearch endpoint, index details, and account credentials. Ensure the account information is valid to avoid task creation failure.

Create a synchronization task

Navigate to the Project List > Project Details > Topic Details page.
In the upper-right corner, click + Sync.
Select Elasticsearch as the connector type.

Parameters

Endpoint

The Elasticsearch service address requires an internal address and an internal port in the format internal address:internal port.

For example, if the internal address is es-cn-xxx.elasticsearch.aliyuncs.com and the internal port is 9200, enter es-cn-xxx.elasticsearch.aliyuncs.com:9200.

Index

You can specify the destination index in two ways: static or dynamic.

Static index

All data is written to a single, predefined index. You must either create this index in advance or enable automatic index creation in Elasticsearch.

Dynamic index

Data is written to different indices based on a time period or the values in a specific data column. You must enable automatic index creation in Elasticsearch for dynamic indexing to work. You can select up to one column to generate the index name.

Supported time formats:

Year	Month	Day	Week
%Y	%m	%d	%U

Example 1: Generate a new index every day at midnight Configure the index as test_${%Y-%m-%d}. If the current date is March 31, 2021, the final index is test_2021-03-31.
Example 2: Generating a new index from a data column The data contains a column named col1, and the index is configured as test_${col1}. If two data records have AAA and BBB as the values for the col1 column, the records are written to the test_AAA and test_BBB indexes respectively.

As the number of indices in Elasticsearch grows, write performance can decrease. An excessive number of indices may cause write timeouts in DataHub. When using a dynamic index, design your naming pattern to avoid creating an excessive number of indices.

User/Password

The username and password for accessing Elasticsearch.

Type column

DataHub generates the document type differently depending on the Elasticsearch version. In Elasticsearch 5, you can create multiple types within a single index. In Elasticsearch 6, you can only create one type per index. This setting cannot be empty.

For Elasticsearch 5, DataHub uses the value of the selected Type column as the document type. If you select multiple columns, their values are concatenated with a "|" separator to form the type. Fields selected for the Type column cannot contain null values.
For Elasticsearch 6, DataHub uses the name of the selected column as the document type. If you select multiple columns, their names are concatenated with a "|" separator. Elasticsearch 6 allows any string as the type name.
You cannot specify a custom type name in the console when creating a synchronization task for Elasticsearch 6. To use a custom type name, you must create the task using the SDK.
For Elasticsearch 7, all documents use a default type, so you do not need to select a Type column.

Example:

DataHub Schema : f1 string, f2 string, f3 string, f4 string
Data record : ["test1","test2","test3",null]

Type column	Elasticsearch 5 type	Elasticsearch 6 type
f1	test1	f1
f1,f3	test1\|test3	f1\|f3
ff	Creation fails	ff
f1,ff	Creation fails	f1\|ff
f4	Creation succeeds, but sync fails (dirty data)	Creation and synchronization succeed

ID column

You can generate a document ID for Elasticsearch from your DataHub record data. If you do not select a column, Elasticsearch generates a unique ID for each document. If you select one or more columns, their values are concatenated with a "|" separator to create the document ID. Fields selected for the ID column cannot contain null values.

Example:

DataHub Schema : f1 string, f2 string, f3 string, f4 string
Data record : ["test1","test2","test3",null]

Id column	Document id
	Elasticsearch auto-generates a unique ID
f1	test1
f1,f3	test1\|test3
ff	Creation fails
f4	Creation succeeds, but sync fails (dirty data)

Router column

You can generate a routing value for Elasticsearch from your DataHub record data. If you do not select a column, the Elasticsearch routing feature is not used. If you select one or more columns, their values are used as the routing value. If you select multiple columns, their values are concatenated with a "|" separator. Fields selected for the Router column cannot contain null values.

For an example, see the ID column.

Imported fields

Select the fields from the DataHub topic to synchronize to Elasticsearch. DataHub does not synchronize unselected fields. For Elasticsearch 5, the final document data does not include the fields used for the ID column and Type column. For more information about how data is transformed, see the examples in the following sections.

Network type

Select the network type based on your Elasticsearch instance deployment. Alibaba Cloud Elasticsearch instances on the public cloud run within a VPC. Therefore, select VPC as the network type for these synchronization tasks. When you use the VPC network type, you must provide the VPC ID and instance ID.

When you enter the instance ID, you must add the suffix -worker. For example, if your instance ID is es-cn-xxx, you must enter es-cn-xxx-worker.

Data writing examples

The following examples assume that the Elasticsearch synchronization task has been created successfully. If task creation fails, review and correct your configuration.

DataHub schema:

Field name	Field type
f1	BIGINT
f2	STRING
f3	BOOLEAN
f4	DOUBLE
f5	TIMESTAMP
f6	DECIMAL

Example 1
- Type column = f1 (Not applicable to Elasticsearch 7)
- ID column = f2
- Imported fields = f1,f2,f3,f4,f5,f6
  - Data record = v1,v2,v3,v4,v5,v6
    ES version
    Type
    Id
    Data
    ES5
    v1
    v2
    {f3:v3,f4:v4,f5:v5,f6:v6}
    ES6
    f1
    v2
    {f1:v1,f3:v3,f4:v4,f5:v5,f6:v6}
    ES7
    -
    v2
    {f1:v1,f3:v3,f4:v4,f5:v5,f6:v6}
  - Data record = null,v2,v3,v4,v5,v6
    ES version
    Type
    Id
    Data
    ES5
    -
    -
    A null value in the Type column results in dirty data.
    ES6
    f1
    v2
    {f1:v1,f3:v3,f4:v4,f5:v5,f6:v6}
    ES7
    -
    v2
    {f1:v1,f3:v3,f4:v4,f5:v5,f6:v6}
  - Data record = v1,null,v3,v4,v5,v6
    ES version
    Type
    Id
    Data
    ES5
    -
    -
    A null value in the ID column results in dirty data.
    ES6
    -
    -
    A null value in the ID column results in dirty data.
    ES7
    -
    -
    A null value in the ID column results in dirty data.
Example 2
- Type column = f1,f2 (Not applicable to Elasticsearch 7)
- ID column = f3,f4
- Imported fields = f1,f2,f3,f4,f5,f6
  - Data record = v1,v2,v3,v4,v5,v6
    ES version
    Type
    Id
    Data
    ES5
    v1|v2
    v3|v4
    {f5:v5,f6:v6}
    ES6
    f1|f2
    v3|v4
    {f1:v1,f2:v2,f5:v5,f6:v6}
    ES7
    -
    v3|v4
    {f1:v1,f2:v2,f5:v5,f6:v6}
  - Data record = v1,null,v3,v4,v5,v6
    ES version
    Type
    Id
    Data
    ES5
    -
    -
    A null value in the Type column results in dirty data.
    ES6
    f1|f2
    v3|v4
    {f1:v1,f2:v2,f5:v5,f6:v6}
    ES7
    -
    v3|v4
    {f1:v1,f2:v2,f5:v5,f6:v6}
  - Data record = v1,v2,null,v4,v5,v6
    ES version
    Type
    Id
    Data
    ES5
    -
    -
    A null value in the ID column results in dirty data.
    ES6
    -
    -
    A null value in the ID column results in dirty data.
    ES7
    -
    -
    A null value in the ID column results in dirty data.

Example 3

Type column = f1 (Not applicable to Elasticsearch 7)
ID column = f2
Router column = f3

Imported fields = f1,f2,f3,f4,f5,f6

Data record = v1,v2,v3,v4,v5,v6
ES version
Type
Id
Router
Data
ES5
v1
v2
v3
{f4:v4,f5:v5,f6:v6}
ES6
f1
v2
v3
{f1:v1,f4:v4,f5:v5,f6:v6}
ES7
-
v2
v3
{f1:v1,f4:v4,f5:v5,f6:v6}
Data record = null,v2,v3,v4,v5,v6
ES version
Type
Id
Data
ES5
-
-
A null value in the Type column results in dirty data.
ES6
f1
v2
{f1:v1,f4:v4,f5:v5,f6:v6}
ES7
-
v2
{f1:v1,f4:v4,f5:v5,f6:v6}
Data record = v1,null,v3,v4,v5,v6
ES version
Type
Id
Data
ES5
-
-
A null value in the ID column results in dirty data.
ES6
-
-
A null value in the ID column results in dirty data.
ES7
-
-
A null value in the ID column results in dirty data.

Data record = v1,v2,null,v4,v5,v6

ES version	Type	Id	Data
ES5	-	-	A null value in the Router column results in dirty data.
ES6	-	-	A null value in the Router column results in dirty data.
ES7	-	-	A null value in the Router column results in dirty data.

Managing the synchronization task

On the details page for the connector, you can view the task's running status, checkpoint, and other information. You can also restart and stop the task. You must stop the task before resetting the checkpoint.

End-to-end example

This example uses Alibaba Cloud Elasticsearch 6.7 to demonstrate the complete process of synchronizing data from DataHub to Elasticsearch. The Elasticsearch-related operations are performed by using Kibana Dev Tools. For other methods, see the official Elasticsearch documentation.

Create an Elasticsearch index
By default, Elasticsearch automatically creates an index, so you can typically skip this step. If automatic index creation is disabled, you must manually create the index. For specific commands, see the official Elasticsearch documentation.
Create a DataHub topic
Only TUPLE-type topics support synchronization to Elasticsearch. For instructions on how to create a DataHub topic, see Topic Operations.
Create the Elasticsearch synchronization task
In this example, when creating the synchronization task, we use f1 and f2 as the Type column, f3 and f4 as the ID column, and select all fields for import.
Write data to the DataHub topic
Use the DataHub SDK or a plugin to write data. After writing a record, you can sample the data in the console to verify what you have written.
Verify the data synchronization
First, check the checkpoint of the Elasticsearch synchronization task. A change in the checkpoint and sync time indicates that data has been synchronized. The sync time reflects when the data was written to DataHub, and a checkpoint value of 1 means that the first data record (at index 0) has been written.
Next, check the data in Elasticsearch. You can use Kibana to confirm that the data has been synchronized successfully.

ES version	Type	Id	Router	Data
ES5	v1	v2	v3	`{f4:v4,f5:v5,f6:v6}`
ES6	f1	v2	v3	`{f1:v1,f4:v4,f5:v5,f6:v6}`
ES7	-	v2	v3	`{f1:v1,f4:v4,f5:v5,f6:v6}`

ES version	Type	Id	Data
ES5	v1\|v2	v3\|v4	`{f5:v5,f6:v6}`
ES6	f1\|f2	v3\|v4	{f1:v1,f2:v2,f5:v5,f6:v6}
ES7	-	v3\|v4	{f1:v1,f2:v2,f5:v5,f6:v6}