Realtime Compute for Apache Flink:Elasticsearch

Write mode: upsert vs. append

The sink table operates in one of two modes, determined by whether a primary key is defined in the DDL.

When a primary key is defined, it is used as the document ID. The connector uses the document-id.key-delimiter parameter (default: _) to join multiple primary key field values into a single document ID string (up to 512 bytes, no spaces), in the order defined in the DDL.

The following types cannot be used as primary key fields because they have no string representation: BYTES, ROW, ARRAY, and MAP.

DDL fields map to Elasticsearch document fields. Metadata such as document IDs is managed by Elasticsearch and cannot be written through the DDL.

Dynamic indexes

The index parameter for sink tables supports both static and dynamic values.

• Static index: A plain string, such as myusers. All records are written to the same index.

• Dynamic index: A template using {field_name} to generate the index name from a field value in each record.

For TIMESTAMP, DATE, and TIME fields, use {field_name|date_format_string} to format the value using Java's DateTimeFormatter. For example, setting index to myusers-{log_ts|yyyy-MM-dd} writes a record with log_ts = 2020-03-27 12:25:55 to the myusers-2020-03-27 index.

Delivery guarantees

When checkpointing is enabled, the connector provides at-least-once delivery by waiting for all pending bulk requests to complete before each checkpoint. If you set sink.flush-on-checkpoint to false, the connector skips this wait and at-least-once delivery guarantees no longer apply.

Source table

CREATE TABLE elasticsearch_source(
  name STRING,
  location STRING,
  value FLOAT
) WITH (
  'connector' = 'elasticsearch',
  'endPoint' = '<yourEndPoint>',
  'indexName' = '<yourIndexName>'
);

Dimension table

CREATE TABLE es_dim(
  field1 STRING,  -- JOIN key; must be STRING type
  field2 FLOAT,
  field3 BIGINT,
  PRIMARY KEY (field1) NOT ENFORCED
) WITH (
  'connector' = 'elasticsearch',
  'endPoint' = '<yourEndPoint>',
  'indexName' = '<yourIndexName>'
);

Primary key behavior for dimension tables:

• With primary key: Only one JOIN key is allowed. The key must match the document ID in the Elasticsearch index.

• Without primary key: One or more JOIN keys are allowed. The keys must be fields in the Elasticsearch document.

For the STRING type, Flink adds a .keyword suffix to field names by default to ensure compatibility. If this prevents matching with Text fields in Elasticsearch, set ignoreKeywordSuffix to true.

Sink table

CREATE TABLE es_sink(
  user_id   STRING,
  user_name STRING,
  uv        BIGINT,
  pv        BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED  -- Optional. If defined, used as the document ID; enables upsert mode.
) WITH (
  'connector' = 'elasticsearch-7',  -- Use elasticsearch-6 for Elasticsearch 6.x, elasticsearch-8 for 8.x (requires VVR 8.0.5+)
  'hosts' = '<yourHosts>',
  'index' = '<yourIndex>'
);

Source table

Sink table

`sink.delete-strategy` options:

Dimension table

Source table

CREATE TEMPORARY TABLE elasticsearch_source (
  name     STRING,
  location STRING,
  `value`  FLOAT
) WITH (
  'connector'  = 'elasticsearch',
  'endPoint'   = '<yourEndPoint>',          -- Example: http://127.0.0.1:XXXX
  'accessId'   = '${secret_values.ak_id}',  -- Store as a project variable
  'accessKey'  = '${secret_values.ak_secret}',
  'indexName'  = '<yourIndexName>',
  'typeNames'  = '<yourTypeName>'           -- Omit for Elasticsearch 7.0 or later
);

CREATE TEMPORARY TABLE blackhole_sink (
  name     STRING,
  location STRING,
  `value`  FLOAT
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT name, location, `value`
FROM elasticsearch_source;

Dimension table

This example joins a generated data stream with an Elasticsearch dimension table using a temporal join.

CREATE TEMPORARY TABLE datagen_source (
  id       STRING,
  data     STRING,
  proctime AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE es_dim (
  id      STRING,
  `value` FLOAT,
  PRIMARY KEY (id) NOT ENFORCED  -- Key must match the document ID in the Elasticsearch index
) WITH (
  'connector' = 'elasticsearch',
  'endPoint'  = '<yourEndPoint>',
  'accessId'  = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'indexName' = '<yourIndexName>',
  'typeNames' = '<yourTypeName>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  id      STRING,
  data    STRING,
  `value` FLOAT
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT e.*, w.*
FROM datagen_source AS e
JOIN es_dim FOR SYSTEM_TIME AS OF e.proctime AS w
ON e.id = w.id;

Sink table — flat schema

CREATE TEMPORARY TABLE datagen_source (
  id   STRING,
  name STRING,
  uv   BIGINT
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE es_sink (
  user_id   STRING,
  user_name STRING,
  uv        BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED  -- Optional. If defined, used as the document ID; enables upsert mode.
) WITH (
  'connector'     = 'elasticsearch-6',           -- Use elasticsearch-7 for Elasticsearch 7.x
  'hosts'         = '<yourHosts>',               -- Example: 127.0.0.1:XXXX
  'index'         = '<yourIndex>',
  'document-type' = '<yourElasticsearch.types>', -- Required for elasticsearch-6 only
  'username'      = '${secret_values.ak_id}',
  'password'      = '${secret_values.ak_secret}'
);

INSERT INTO es_sink
SELECT id, name, uv
FROM datagen_source;

Sink table — nested schema

This example writes data with a nested ROW type, including ARRAY and MAP fields.

CREATE TEMPORARY TABLE datagen_source (
  id      STRING,
  details ROW<
    name       STRING,
    ages       ARRAY<INT>,
    attributes MAP<STRING, STRING>
  >
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE es_sink (
  id      STRING,
  details ROW<
    name       STRING,
    ages       ARRAY<INT>,
    attributes MAP<STRING, STRING>
  >,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector'     = 'elasticsearch-6',
  'hosts'         = '<yourHosts>',
  'index'         = '<yourIndex>',
  'document-type' = '<yourElasticsearch.types>',
  'username'      = '${secret_values.ak_id}',
  'password'      = '${secret_values.ak_secret}'
);

INSERT INTO es_sink
SELECT id, details
FROM datagen_source;