Sink data from RisingWave to Elasticsearch

This guide describes how to sink data from RisingWave to Elasticsearch using the Elasticsearch sink connector in RisingWave.

Elasticsearch is a distributed, RESTful search and analytics engine capable of addressing a growing number of use cases. It centrally stores your data for lightning-fast search, fine‑tuned relevancy, and powerful analytics that scale with ease.

The Elasticsearch sink connecter in RisingWave will perform index operations via the bulk API, flushing updates whenever one of these criteria is met:

• 1,000 operations
• 5mb of updates
• 5 seconds since the last flush (assuming new actions are queued)

​

Prerequisites

• Ensure the Elasticsearch cluster (version 7.x or 8.x) is accessible from RisingWave.
• If you are running RisingWave locally from binaries, make sure that you have JDK 11 or later versions installed in your environment.

​

Create an Elasticsearch sink

Use the following syntax to create an Elasticsearch sink. Once a sink is created, any insert or update to the sink will be streamed to the specified Elasticsearch endpoint.

CREATE SINK sink_name
[ FROM sink_from | AS select_query ]
WITH (
  connector = 'elasticsearch',
  primary_key = '<primary key of the sink_from object>',
  { index = '<your Elasticsearch index>' | index_column = '<your index column>' },
  url = 'http://<ES hostname>:<ES port>',
  username = '<your ES username>',
  password = '<your password>',
  delimiter='<delimiter>'
);

​

Parameters

So, if you are using Elasticsearch 7.x, we set it to the official’s recommended value, which is ‘_doc’. If you are using Elasticsearch 8.x, this parameter has been removed by the Elasticsearch official, so no setting is required.

​

Notes about primary keys and Elasticsearch IDs

The Elasticsearch sink defaults to the upsert sink type. It does not support the append-only sink type.

If you want to customize your Elasticsearch ID, please specify it via the primary_key parameter. RisingWave will combine multiple primary key values into a single string with the delimiter you set, and use it as the Elasticsearch ID.

If you don’t want to customize your Elasticsearch ID, RisingWave will use the first column in the sink definition as the Elasticsearch ID.

​

Data type mapping

ElasticSearch uses a mechanism called dynamic field mapping to dynamically create fields and determine their types automatically. It treats all integer types as long and all floating-point types as float. To ensure data types in RisingWave are mapped to the data types in Elasticsearch correctly, we recommend that you specify the mapping via index templates or dynamic templates before creating the sink.