ElasticからClickStackへのデータ移行

パラレルオペレーション戦略

ElasticからClickStackへの移行を行う際、特に可観測性のユースケースでは、過去のデータを移行する代わりにパラレルオペレーションアプローチを推奨します。この戦略にはいくつかの利点があります。

1. 最小限のリスク: 並行して両方のシステムを運用することで、既存のデータやダッシュボードへのアクセスを維持しながら、ClickStackを検証し、新しいシステムにユーザーを慣れさせることができます。
2. 自然なデータの有効期限: ほとんどの可観測性データは、限られた保持期間（通常30日以下）を持ち、Elasticからデータが期限切れになるにつれて、自然な移行が可能です。
3. 簡略化された移行: システム間で歴史的なデータを移動させるための複雑なデータ転送ツールやプロセスは不要です。

データ移行

ElasticsearchからClickHouseに重要なデータを移行するためのアプローチは、セクション"データ移行"で示しています。この方法は、大規模なデータセットには推奨されません。なぜなら、Elasticsearchが効率的にエクスポートする能力に制限があり、JSON形式のみがサポートされているためです。

実装ステップ

1. デュアルインジェスト設定

データコレクションパイプラインを設定して、ElasticとClickStackの両方に同時にデータを送信します。

これを達成する方法は、現在使用している収集エージェントに依存します。詳細は"エージェントの移行"を参照してください。

2. 保持期間の調整

ElasticのTTL設定を希望する保持期間に合わせて構成します。ClickStackのTTLを設定して、同じ期間データを保持します。

3. 検証と比較:

• 両方のシステムに対してクエリを実行し、データの整合性を確認します
• クエリのパフォーマンスと結果を比較します
• ダッシュボードとアラートをClickStackに移行します。これは現在手動のプロセスです。
• 重要なダッシュボードとアラートがClickStackで期待通りに機能することを確認します

4. 段階的移行:

• データがElasticから自然に期限切れになるにつれ、ユーザーはますますClickStackに依存するようになります
• ClickStackへの信頼が確立され次第、クエリとダッシュボードのリダイレクトを開始できます

長期保持

長期の保持期間を必要とする組織の場合:

• 全てのデータがElasticから期限切れになるまで、両方のシステムを並行して運用し続けます
• ClickStackの階層ストレージ機能を活用して、長期データを効率的に管理できます。
• アグリゲートまたはフィルタリングされた歴史データを保持しながら、生データを期限切れにするためにマテリアライズドビューの使用を検討してください。

移行タイムライン

移行のタイムラインは、データ保持要件に依存します:

• 30日保持: 移行は1か月以内に完了できます。
• 長期保持: データがElasticから期限切れになるまで並行運用を続けます。
• 歴史データ: 絶対必要な場合は、特定の歴史データをインポートするためにデータ移行の使用を検討してください。

移行設定

ElasticからClickStackに移行する際には、インデックスとストレージの設定をClickHouseのアーキテクチャに合わせて調整する必要があります。Elasticsearchはパフォーマンスと障害耐性のために水平スケーリングとシャーディングに依存しているため、デフォルトで複数のシャードを持っていますが、ClickHouseは垂直スケーリングに最適化されており、通常は少ないシャードで最高のパフォーマンスを発揮します。

推奨設定

単一シャードから始め、垂直にスケーリングすることを推奨します。この構成は、ほとんどの可観測性ワークロードに適しており、管理とクエリパフォーマンスのチューニングを簡素化します。

• ClickHouse Cloud: デフォルトで単一シャード、マルチレプリカアーキテクチャを使用します。ストレージとコンピュートは独立してスケールし、予測不可能なインジェストパターンおよび読み取り重視のワークロードに理想的です。
• ClickHouse OSS: セルフマネージドデプロイでは、次のことを推奨します:
  • 単一シャードから開始する
  • 追加のCPUとRAMで垂直にスケーリングする
  • 階層ストレージを用いてローカルディスクをS3互換のオブジェクトストレージで拡張する
  • 高可用性が必要な場合はReplicatedMergeTreeを使用する
  • 障害耐性のために、可観測性ワークロードでは1つのレプリカが通常十分です。

シャーディングのタイミング

シャーディングが必要になる場合:

• インジェストレートが単一ノードの容量を超える（通常は500K行/秒を超える）
• テナントの隔離や地域データの分離が必要
• 全データセットが単一のサーバーでは大きすぎる（オブジェクトストレージを使用しても）

シャーディングが必要な場合は、シャードキーと分散テーブルのセットアップに関するガイダンスは水平スケーリングを参照してください。

保持とTTL

ClickHouseは、MergeTreeテーブル上でデータの有効期限管理のためにTTL句を使用します。TTLポリシーは次のことができます：

• 期限切れデータを自動的に削除する
• 古いデータをコールドオブジェクトストレージに移動する
• 最近の頻繁にクエリされるログのみを高速ディスクに保持する

移行中のデータライフサイクルを維持するために、ClickHouseのTTL設定を既存のElasticの保持ポリシーに合わせることを推奨します。例として、ClickStackの本番TTL設定を参照してください。

データ移行

ほとんどの可観測性データにはパラレルオペレーションを推奨していますが、ElasticからClickHouseへの直接的なデータ移行が必要な特定のケースもあります。

• データエンリッチメントに使用される小規模なルックアップテーブル（例：ユーザーのマッピング、サービスカタログ）
• 可観測性データと相関させる必要があるElasticに保存されたビジネスデータ。この場合、ClickHouseのSQL機能とビジネスインテリジェンス統合により、Elasticのより制限されたクエリオプションと比較して、データの保持とクエリが容易になります。
• 移行中に保持する必要のある設定データ

このアプローチは、データセットが1000万行未満の場合にのみ有効です。なぜなら、Elasticsearchのエクスポート能力はHTTP経由でのJSONに制限されており、大規模なデータセットに対してはスケールしないためです。

以下のステップでは、ClickHouseからElasticの単一インデックスを移行することが可能です。

スキーマの移行

Elasticsearchから移行するインデックス用にClickHouseにテーブルを作成します。ユーザーは、Elasticsearchの型をClickHouseの等価物にマッピングすることができます。あるいは、ユーザーはClickHouseのJSONデータ型に単純に依存して、データが挿入されるときに適切な型のカラムを動的に作成することもできます。

以下はsyslogデータを含むインデックスに対するElasticsearchのマッピングです。

Elasticsearchマッピング

GET .ds-logs-system.syslog-default-2025.06.03-000001/_mapping
{
  ".ds-logs-system.syslog-default-2025.06.03-000001": {
    "mappings": {
      "_meta": {
        "managed_by": "fleet",
        "managed": true,
        "package": {
          "name": "system"
        }
      },
      "_data_stream_timestamp": {
        "enabled": true
      },
      "dynamic_templates": [],
      "date_detection": false,
      "properties": {
        "@timestamp": {
          "type": "date",
          "ignore_malformed": false
        },
        "agent": {
          "properties": {
            "ephemeral_id": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "id": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "name": {
              "type": "keyword",
              "fields": {
                "text": {
                  "type": "match_only_text"
                }
              }
            },
            "type": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "version": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "cloud": {
          "properties": {
            "account": {
              "properties": {
                "id": {
                  "type": "keyword",
                  "ignore_above": 1024
                }
              }
            },
            "availability_zone": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "image": {
              "properties": {
                "id": {
                  "type": "keyword",
                  "ignore_above": 1024
                }
              }
            },
            "instance": {
              "properties": {
                "id": {
                  "type": "keyword",
                  "ignore_above": 1024
                }
              }
            },
            "machine": {
              "properties": {
                "type": {
                  "type": "keyword",
                  "ignore_above": 1024
                }
              }
            },
            "provider": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "region": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "service": {
              "properties": {
                "name": {
                  "type": "keyword",
                  "fields": {
                    "text": {
                      "type": "match_only_text"
                    }
                  }
                }
              }
            }
          }
        },
        "data_stream": {
          "properties": {
            "dataset": {
              "type": "constant_keyword",
              "value": "system.syslog"
            },
            "namespace": {
              "type": "constant_keyword",
              "value": "default"
            },
            "type": {
              "type": "constant_keyword",
              "value": "logs"
            }
          }
        },
        "ecs": {
          "properties": {
            "version": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "elastic_agent": {
          "properties": {
            "id": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "snapshot": {
              "type": "boolean"
            },
            "version": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "event": {
          "properties": {
            "agent_id_status": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "dataset": {
              "type": "constant_keyword",
              "value": "system.syslog"
            },
            "ingested": {
              "type": "date",
              "format": "strict_date_time_no_millis||strict_date_optional_time||epoch_millis",
              "ignore_malformed": false
            },
            "module": {
              "type": "constant_keyword",
              "value": "system"
            },
            "timezone": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "host": {
          "properties": {
            "architecture": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "containerized": {
              "type": "boolean"
            },
            "hostname": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "id": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "ip": {
              "type": "ip"
            },
            "mac": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "name": {
              "type": "keyword",
              "ignore_above": 1024
            },
            "os": {
              "properties": {
                "build": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "codename": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "family": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "kernel": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "name": {
                  "type": "keyword",
                  "fields": {
                    "text": {
                      "type": "match_only_text"
                    }
                  }
                },
                "platform": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "type": {
                  "type": "keyword",
                  "ignore_above": 1024
                },
                "version": {
                  "type": "keyword",
                  "ignore_above": 1024
                }
              }
            }
          }
        },
        "input": {
          "properties": {
            "type": {
              "type": "keyword",
              "ignore_above": 1024
            }
          }
        },
        "log": {
          "properties": {
            "file": {
              "properties": {
                "path": {
                  "type": "keyword",
                  "fields": {
                    "text": {
                      "type": "match_only_text"
                    }
                  }
                }
              }
            },
            "offset": {
              "type": "long"
            }
          }
        },
        "message": {
          "type": "match_only_text"
        },
        "process": {
          "properties": {
            "name": {
              "type": "keyword",
              "fields": {
                "text": {
                  "type": "match_only_text"
                }
              }
            },
            "pid": {
              "type": "long"
            }
          }
        },
        "system": {
          "properties": {
            "syslog": {
              "type": "object"
            }
          }
        }
      }
    }
  }
}

同等のClickHouseテーブルスキーマ：

ClickHouseスキーマ

SET enable_json_type = 1;

CREATE TABLE logs_system_syslog
(
    `@timestamp` DateTime,
    `agent` Tuple(
        ephemeral_id String,
        id String,
        name String,
        type String,
        version String),
    `cloud` Tuple(
        account Tuple(
            id String),
        availability_zone String,
        image Tuple(
            id String),
        instance Tuple(
            id String),
        machine Tuple(
            type String),
        provider String,
        region String,
        service Tuple(
            name String)),
    `data_stream` Tuple(
        dataset String,
        namespace String,
        type String),
    `ecs` Tuple(
        version String),
    `elastic_agent` Tuple(
        id String,
        snapshot UInt8,
        version String),
    `event` Tuple(
        agent_id_status String,
        dataset String,
        ingested DateTime,
        module String,
        timezone String),
    `host` Tuple(
        architecture String,
        containerized UInt8,
        hostname String,
        id String,
        ip Array(Variant(IPv4, IPv6)),
        mac Array(String),
        name String,
        os Tuple(
            build String,
            codename String,
            family String,
            kernel String,
            name String,
            platform String,
            type String,
            version String)),
    `input` Tuple(
        type String),
    `log` Tuple(
        file Tuple(
            path String),
        offset Int64),
    `message` String,
    `process` Tuple(
        name String,
        pid Int64),
    `system` Tuple(
        syslog JSON)
)
ENGINE = MergeTree
ORDER BY (`host.name`, `@timestamp`)

注意点：

• ネストされた構造はドット表記の代わりにタプルを使用して表します
• マッピングに基づいて適切なClickHouse型を使用しています：
  • keyword → String
  • date → DateTime
  • boolean → UInt8
  • long → Int64
  • ip → Array(Variant(IPv4, IPv6))。ここではVariant(IPv4, IPv6)を使用しています、なぜならフィールドにはIPv4とIPv6といった混合が含まれているためです。
  • object → 構造が不確実なsyslogオブジェクトのためにJSONを使用します。
• カラム host.ip と host.mac は、Elasticsearchで全ての型が配列であるのに対し、明示的なArray型です。
• タイムスタンプとホスト名を使用して、効率的な時間ベースのクエリのためにORDER BY句が追加されています
• ログデータに最適なMergeTreeがエンジンタイプとして使用されます

このスキーマを静的に定義し、必要に応じてJSON型を選択的に使用するアプローチは推奨されます。

この厳密なスキーマには多くの利点があります：

• データ検証 – 厳密なスキーマを強制することで、特定の構造外でのカラムの爆発のリスクを回避します。
• カラム爆発のリスクを回避: JSON型は潜在的に何千ものカラムにスケールする可能性がありますが、サブカラムが専用のカラムとして保存されていると、極端に多くのカラムファイルが作成され、パフォーマンスに影響を与えるカラムファイルの爆発を引き起こすことがあります。これを軽減するために、JSONによって使用される基盤となる動的型は、別のカラムファイルとして保存されるユニークパスの数を制限するmax_dynamic_pathsパラメータを提供します。閾値に達すると、追加のパスはコンパクトにエンコードされた形式の共有カラムファイルに保存され、パフォーマンスとストレージの効率が維持されながら柔軟なデータインジェストがサポートされます。ただし、この共有カラムファイルにアクセスすることは、パフォーマンスがそれほど良くはありません。JSONカラムは、型ヒントと共に使用されることもありますので注意してください。「ヒント付き」カラムは、専用カラムと同じパフォーマンスを提供します。
• パスと型の簡単な内省: JSON型は、内省関数をサポートしており、推測された型およびパスを特定できますが、静的構造は、例えばDESCRIBEを使ってより簡単に探ることができます。

あるいは、ユーザーは単純にJSONカラムを1つ持つテーブルを作成することができます。

SET enable_json_type = 1;

CREATE TABLE syslog_json
(
 `json` JSON(`host.name` String, `@timestamp` DateTime)
)
ENGINE = MergeTree
ORDER BY (`json.host.name`, `json.@timestamp`)

注記

host.nameとtimestampカラムの型ヒントをJSON定義に提供します。これを使用して、整列/主キーに使います。これにより、ClickHouseはこのカラムがnullではないことを知って、どのサブカラムを使うべきかを把握できます（各型には複数あるかもしれないので、そうでなければあいまいです）。

この後者のアプローチは、単純ですが、プロトタイピングやデータエンジニアリングのタスクに最適です。本番環境では、必要な動的サブ構造についてのみJSONを使用してください。

スキーマ内でのJSON型の使用、および効率的に適用する方法に関する詳細は、"スキーマの設計"ガイドを推奨します。

elasticdumpのインストール

Elasticsearchからデータをエクスポートするためにelasticdumpを推奨します。このツールはnodeを必要とし、ElasticsearchとClickHouseの両方にネットワーク的に近いマシンにインストールする必要があります。ほとんどのエクスポートに対して、少なくとも4コアおよび16GBのRAMを持つ専用サーバーを推奨します。

npm install elasticdump -g

elasticdumpはデータ移行にいくつかの利点を提供します：

• ElasticsearchのREST APIと直接やり取りし、適切なデータエクスポートを保証します。
• エクスポートプロセス中にPoint-in-Time (PIT) APIを使用してデータの整合性を維持します。これにより特定の瞬間のデータの一貫したスナップショットを作成します。
• データをJSON形式で直接エクスポートし、ClickHouseクライアントにストリーミングして挿入できます。

可能であれば、ClickHouse、Elasticsearch、およびelasticdumpを同じアベイラビリティゾーンまたはデータセンターで実行し、ネットワークの出力を最小限に抑え、スループットを最大化することを推奨します。

ClickHouseクライアントのインストール

elasticdumpが存在するサーバーにClickHouseがインストールされていることを確認してください。ClickHouseサーバーを起動しないでください - これらのステップはクライアントのみが必要です。

データのストリーミング

ElasticsearchとClickHouseの間でデータをストリーミングするには、elasticdumpコマンドを使用し、出力を直接ClickHouseクライアントにパイプします。以下はデータを、きちんと構造化されたテーブルlogs_system_syslogに挿入します。

# 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"

elasticdumpで使用される次のフラグに注意してください：

• type=data - Elasticsearchにおける文書コンテンツのみへの応答を制限します。
• input-index - Elasticsearchの入力インデックス。
• output=$ - すべての結果をstdoutにリダイレクトします。
• メタデータフィールドを応答から省略することを保証するsourceOnlyフラグ。
• 結果の効率的なページネーションのためにsearchAfter APIを使用するsearchAfterフラグ。
• ポイント・イン・タイムAPIを使用してクエリ間で一貫した結果を保証するpit=true。

ここでのClickHouseクライアントのパラメータ（資格情報以外）：

• max_insert_block_size=1000 - ClickHouseクライアントは、この行数に達するとデータを送信します。数値を増加させることでスループットが改善されますが、ブロックを形成する時間が増加するため、ClickHouseにデータが表示されるまでの時間が長くなります。
• min_insert_block_size_bytes=0 - バイトによるサーバーブロックスクワッシングを無効にします。
• min_insert_block_size_rows=1000 - サーバー側でクライアントからのブロックをスカッシュします。この場合、max_insert_block_sizeに設定して、行が即座に表示されるようにします。スループットを改善するためには数値を増加させます。
• query="INSERT INTO logs_system_syslog FORMAT JSONAsRow" - データをJSONEachRow形式で挿入します。これはlogs_system_syslogのような明確に定義されたスキーマに送信する場合に適しています。

ユーザーは毎秒数千行のスループットを期待できます。

単一JSON行への挿入

単一JSONカラムに挿入する場合（上記のsyslog_jsonスキーマを参照）、同じ挿入コマンドを使用できますが、ユーザーは形式をJSONAsObjectとして指定する必要があります。例：

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"

詳細については"オブジェクトとしてJSONを読む"を参照してください。

データの変換（オプション）

上記のコマンドは、ElasticsearchフィールドとClickHouseカラム間の1:1マッピングを前提としています。しかし、ユーザーはElasticsearchデータをClickHouseに挿入する前にフィルターや変換が必要なことがよくあります。

これは、SELECTクエリをstdout上で実行できるinputテーブル関数を使用すると実現できます。

例えば、以前のデータからtimestampとhostnameフィールドだけを保存したいとします。ClickHouseのスキーマ：

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

elasticdumpからこのテーブルに挿入するためには、JSON型を使用して必要なカラムを動的に検出し選択することで、inputテーブル関数を単純に使用できます。注：このSELECTクエリにはフィルタを含むことができます。

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"

@timestampフィールド名をエスケープする必要があり、入力形式としてJSONAsObjectを使用することに注意してください。