跳到主要内容
版本:2.1.3

Flink SQL ElasticSearch Connector

Description

With elasticsearch connector, you can use the Flink SQL to write data into ElasticSearch.

Usage

Let us have a brief example to show how to use the connector.

1. kafka prepare

Please refer to the Eleastic Doc to prepare elastic environment.

2. prepare seatunnel configuration

ElasticSearch provide different connectors for different version:

  • version 6.x: flink-sql-connector-elasticsearch6
  • version 7.x: flink-sql-connector-elasticsearch7

Here is a simple example of seatunnel configuration.

SET table.dml-sync = true;

CREATE TABLE events (
id INT,
name STRING
) WITH (
'connector' = 'datagen'
);

CREATE TABLE es_sink (
id INT,
name STRING
) WITH (
'connector' = 'elasticsearch-7', -- or 'elasticsearch-6'
'hosts' = 'http://localhost:9200',
'index' = 'users'
);

INSERT INTO es_sink SELECT * FROM events;

Execute the following command in seatunnel home path to start the Flink SQL job.

$ bin/start-seatunnel-sql.sh -c config/elasticsearch.sql.conf

4. verify result

Verify result from elasticsearch.