Elasticsearch

Elasticsearch sink connector

Description

Output data to Elasticsearch.

tip

Engine Supported and plugin name

• Spark: Elasticsearch(supported ElasticSearch version is >= 2.x and <7.0.0)
• Flink: Elasticsearch(supported ElasticSearch version = 7.x, if you want use Elasticsearch version is 6.x, please use the source code to repackage by execute mvn clean package -Delasticsearch=6)

Options

Spark

name	type	required	default value
hosts	array	yes	-
index_type	string	no	-
index_time_format	string	no	yyyy.MM.dd
index	string	no	seatunnel
es.*	string	no
common-options	string	no	-

Flink

name	type	required	default value
hosts	array	yes	-
index_type	string	no	log
index_time_format	string	no	yyyy.MM.dd
index	string	no	seatunnel
common-options	string	no	-
parallelism	int	no	-

hosts [array]

Elasticsearch cluster address, the format is host:port , allowing multiple hosts to be specified. Such as ["host1:9200", "host2:9200"] .

index_type [string]

Elasticsearch index type, it is recommended not to specify in elasticsearch 7 and above

index_time_format [string]

When the format in the index parameter is xxxx-${now} , index_time_format can specify the time format of the index name, and the default value is yyyy.MM.dd . The commonly used time formats are listed as follows:

Symbol	Description
y	Year
M	Month
d	Day of month
H	Hour in day (0-23)
m	Minute in hour
s	Second in minute

See Java SimpleDateFormat for detailed time format syntax.

index [string]

Elasticsearch index name. If you need to generate an index based on time, you can specify a time variable, such as seatunnel-${now} . now represents the current data processing time.

Spark

es.* [string]

Users can also specify multiple optional parameters. For a detailed list of parameters, see Parameters Supported by Elasticsearch.

For example, the way to specify es.batch.size.entries is: es.batch.size.entries = 100000 . If these non-essential parameters are not specified, they will use the default values given in the official documentation.

Flink

parallelism [Int]

The parallelism of an individual operator, data source, or data sink

common options [string]

Sink plugin common parameters, please refer to Sink Plugin for details

Examples

elasticsearch {
    hosts = ["localhost:9200"]
    index = "seatunnel"
}