Elasticsearch sink

The Elasticsearch sink writes change data into an Elasticsearch index via the Elasticsearch REST API.

​

Configuration

• Endpoint URL

  Base URL of your Elasticsearch cluster (for example, https://your-es-server:9200).

• Index name

  Name of the Elasticsearch index to write to. The index must exist before Sequin can stream data.

• Authentication type

  One of api_key, basic, or bearer.

• Authentication value

  Credential corresponding to the chosen authentication type.

• Batch size (optional)

  Maximum number of documents per bulk request (default 100, maximum 10 000).

​

Authentication header formats

Sequin constructs the header automatically—supply only the value shown in angle brackets.

​

Transform requirements

Your transform must return a JSON document compatible with the target index’s mapping.

Sequin sets the Elasticsearch document _id to the concatenation of the primary‑key column values of the Postgres row. Let us know if you need to use a different _id field.

Because the sink always uses the index bulk operation, every payload must contain the full document. Partial updates are not supported.

​

Example transforms

Index the full document:

def transform(_action, record, _changes, _metadata) do
  record
end

Index certain fields, renaming from Postgres column names to Elasticsearch field names:

def transform(_action, record, _changes, _metadata) do
  %{
    id: record["product_id"],
    name: record["product_name"],
    description: record["product_description"]
  }
end

​

Bulk import behaviour

For each batch Sequin posts a request to the Elasticsearch Bulk API:

POST /{index_name}/_bulk

All Bulk API payloads are newline‑delimited JSON (NDJSON) and end with a newline as required by Elasticsearch.

​

API endpoints used by Sequin

• POST /{index_name}/_bulk – indexing, replacing, and deleting documents (always used, even for a single document).
• POST /{index_name}/_search – invoked by Test Connection in the console to verify connectivity.

No other Elasticsearch endpoints are called.

​

Error handling

Typical errors surfaced in the Sequin console include:

• Connection or TLS failures.
• 404 – index not found.
• 401 / 403 – authentication or authorisation failure.
• Mapping conflicts (field type mismatches).
• Malformed bulk payload (usually due to transform output).

Errors from Elasticsearch are shown verbatim in the Messages tab. For bulk requests Sequin maps individual item failures to each corresponding document in the batch.

​

Limits

• Batch size is limited to 10,000 documents by Sequin. Elasticsearch’s default http.max_content_length (100 MB) may require smaller batches.
• One sink targets one index; multiple indices require multiple sinks.
• Only index and delete bulk operations are issued. Other operations (update, create, etc.) are not used.

​

Routing

The Elasticsearch sink supports dynamic routing of the index_name with routing functions.

Example routing function:

def route(action, record, changes, metadata) do
  %{
    index_name: metadata.table_name
  }
end

When not using a routing function, messages will be indexed into the statically configured index.