Migrating data to ClickStack from Elastic

```
# export url and credentials
export ELASTICSEARCH_INDEX=.ds-logs-system.syslog-default-2025.06.03-000001
export ELASTICSEARCH_URL=
export ELASTICDUMP_INPUT_USERNAME=
export ELASTICDUMP_INPUT_PASSWORD=
export CLICKHOUSE_HOST=
export CLICKHOUSE_PASSWORD=
export CLICKHOUSE_USER=default

# command to run - modify as required
elasticdump --input=${ELASTICSEARCH_URL} --type=data --input-index ${ELASTICSEARCH_INDEX} --output=$ --sourceOnly --searchAfter --pit=true | 
clickhouse-client --host ${CLICKHOUSE_HOST} --secure --password ${CLICKHOUSE_PASSWORD} --user ${CLICKHOUSE_USER} --max_insert_block_size=1000 \
--min_insert_block_size_bytes=0 --min_insert_block_size_rows=1000 --query="INSERT INTO test.logs_system_syslog FORMAT JSONEachRow"
```

Note the use of the following flags for `elasticdump`:

- `type=data` - limits the response to only the document content in Elasticsearch.
- `input-index` - our Elasticsearch input index.
- `output=$` - redirects all results to stdout.
- `sourceOnly` flag ensuring we omit metadata fields in our response.
- `searchAfter` flag to use the [`searchAfter` API](https://www.elastic.co/docs/reference/elasticsearch/rest-apis/paginate-search-results#search-after) for efficient pagination of results.
- `pit=true` to ensure consistent results between queries using the [point in time API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-open-point-in-time).

  

Our ClickHouse client parameters here (aside from credentials):

- `max_insert_block_size=1000` - ClickHouse client will send data once this number of rows is reached. Increasing improves throughput at the expense of time to formulate a block - thus increasing time till data appears in ClickHouse.
- `min_insert_block_size_bytes=0` - Turns off server block squashing by bytes.
- `min_insert_block_size_rows=1000` - Squashes blocks from clients on the server side. In this case, we set to `max_insert_block_size` so rows appear immediately. Increase to improve throughput.
- `query="INSERT INTO logs_system_syslog FORMAT JSONAsRow"` - Inserting the data as [JSONEachRow format](/docs/integrations/data-formats/json/other-formats). This is appropriate if sending to a well-defined schema such as `logs_system_syslog.`

  

**You can expect throughput on the order of thousands of rows per second.**

Inserting into single JSON row

If inserting into a single JSON column (see the `syslog_json` schema above), the same insert command can be used. However, you must specify `JSONAsObject` as the format instead of `JSONEachRow` e.g.

```
elasticdump --input=${ELASTICSEARCH_URL} --type=data --input-index ${ELASTICSEARCH_INDEX} --output=$ --sourceOnly --searchAfter --pit=true | 
clickhouse-client --host ${CLICKHOUSE_HOST} --secure --password ${CLICKHOUSE_PASSWORD} --user ${CLICKHOUSE_USER} --max_insert_block_size=1000 \
--min_insert_block_size_bytes=0 --min_insert_block_size_rows=1000 --query="INSERT INTO test.logs_system_syslog FORMAT JSONAsObject"
```

See ["Reading JSON as an object"](/docs/integrations/data-formats/json/other-formats#reading-json-as-an-object) for further details.

Transform data (optional)

### Transform data (optional)[​](#transform-data "Direct link to Transform data (optional)")

The above commands assume a 1:1 mapping of Elasticsearch fields to ClickHouse columns. Users often need to filter and transform Elasticsearch data before insertion into ClickHouse.

This can be achieved using the [`input`](/docs/sql-reference/table-functions/input) table function, which allows us to execute any `SELECT` query on the stdout.

Suppose we wish to only store the `timestamp` and `hostname` fields from our earlier data. The ClickHouse schema:

```
CREATE TABLE logs_system_syslog_v2
(
    `timestamp` DateTime,
    `hostname` String
)
ENGINE = MergeTree
ORDER BY (hostname, timestamp)
```

To insert from `elasticdump` into this table, we can simply use the `input` table function - using the JSON type to dynamically detect and select the required columns. Note this `SELECT` query could easily contain a filter.

```
elasticdump --input=${ELASTICSEARCH_URL} --type=data --input-index ${ELASTICSEARCH_INDEX} --output=$ --sourceOnly --searchAfter --pit=true |
clickhouse-client --host ${CLICKHOUSE_HOST} --secure --password ${CLICKHOUSE_PASSWORD} --user ${CLICKHOUSE_USER} --max_insert_block_size=1000 \
--min_insert_block_size_bytes=0 --min_insert_block_size_rows=1000 --query="INSERT INTO test.logs_system_syslog_v2 SELECT json.\`@timestamp\` as timestamp, json.host.hostname as hostname FROM input('json JSON') FORMAT JSONAsObject"
```

Note the need to escape the `@timestamp` field name and use the `JSONAsObject` input format.

- [Parallel operation strategy](#parallel-operation-strategy)
  - [Implementation steps](#implementation-steps)- [Long-term retention](#long-term-retention)- [Migration timeline](#migration-timeline)- [Migrating settings](#migration-settings)
    - [Recommended settings](#recommended-settings)- [When to shard](#when-to-shard)- [Retention and TTL](#retention-and-ttl)- [Migrating data](#migrating-data)
      - [Migrate schema](#migrate-scheme)- [Install `elasticdump`](#install-elasticdump)- [Install ClickHouse client](#install-clickhouse-client)- [Stream data](#stream-data)- [Transform data (optional)](#transform-data)

Was this page helpful?