Elastic から ClickStack へのデータ移行

並行運用戦略

オブザーバビリティ用途で Elastic から ClickStack へ移行する際には、履歴データの移行を試みるのではなく、並行運用 アプローチを推奨します。この戦略には次の利点があります。

1. リスクの最小化: 両方のシステムを同時に稼働させることで、ClickStack を検証しつつ、新しいシステムにユーザーが慣れる間も既存のデータとダッシュボードへのアクセスを維持できます。
2. 自然なデータの有効期限切れ: ほとんどのオブザーバビリティデータは保持期間が限られており（通常 30 日以内）、Elastic 上のデータが有効期限切れとなるにつれて自然な形で移行を進めることができます。
3. 移行の単純化: システム間で履歴データを移動するための複雑なデータ転送ツールやプロセスが不要になります。

データ移行

"Migrating data" セクションでは、Elasticsearch から ClickHouse に重要なデータを移行するためのアプローチを紹介します。これは、大規模なデータセットにはほとんどの場合パフォーマンス上適していません。Elasticsearch 側のエクスポート性能に制約されるうえ、サポートされる形式が JSON のみであるためです。

実装手順

1. 二重インジェストを構成する

データ収集パイプラインを設定し、Elastic と ClickStack の両方に同時にデータを送信できるようにします。

これをどのように実現するかは、現在使用している収集用エージェントによって異なります。詳しくは、「Migrating Agents」を参照してください。

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 行/秒を超える場合）
• テナント分離やリージョンごとのデータ分離が必要な場合
• オブジェクトストレージを利用しても、単一サーバーに収まりきらないほどデータセットが大きい場合

シャーディングが必要な場合は、シャードキーと分散テーブル構成に関するガイダンスとして、Horizontal scaling を参照してください。

保持期間と TTL

ClickHouse は MergeTree テーブルの TTL 句 を使用して、データの有効期限を管理します。TTL ポリシーにより、次のことが可能です。

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

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

データの移行

ほとんどのオブザーバビリティデータについては併用運用を推奨しますが、Elasticsearch から ClickHouse へデータを直接移行する必要があるケースもあります。

• データエンリッチメントに使用される小さなルックアップテーブル（例：ユーザーマッピング、サービスカタログ）
• オブザーバビリティデータと相関付ける必要がある、Elasticsearch に保存されたビジネスデータ。ClickHouse の SQL 機能と Business Intelligence との連携により、クエリ機能が制限されている Elasticsearch と比べて、これらのデータの保守およびクエリが容易になります。
• 移行前後で保持する必要がある構成データ

このアプローチが実用的なのは、1,000 万行未満のデータセットに限られます。これは、Elasticsearch のエクスポート機能が HTTP 経由の JSON に限られており、大規模なデータセットにはうまくスケールしないためです。

以下の手順では、単一の Elasticsearch インデックスを ClickHouse に移行します。

スキーマの移行

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))。このフィールドには IPv4 と IPv6 が混在しているため、Variant(IPv4, IPv6) を使用します。
  • object → 構造が予測できない syslog オブジェクトには JSON を使用します。
• host.ip と host.mac の列は明示的に Array 型として定義されています。これは、すべての型が暗黙的に配列として扱われる Elasticsearch とは異なります。
• タイムスタンプとホスト名をキーとする ORDER BY 句を追加し、時間ベースのクエリを効率化します
• ログデータに最適な MergeTree がエンジン種別として使用されます

スキーマを静的に定義し、必要な箇所でJSON型を選択的に使用するこのアプローチが推奨されます。

この厳密なスキーマには、次のような利点があります：

• データ検証 – 厳密なスキーマを採用することで、特定の構造を除き、カラム爆発のリスクを回避できます。
• 列の爆発的増加のリスクを回避: JSON 型ではサブカラムが専用のカラムとして保存されるため、潜在的には数千のカラムまでスケールできますが、その結果として過剰な数のカラムファイルが生成され、パフォーマンスに悪影響を及ぼす「カラムファイルの爆発」を引き起こす可能性があります。これを緩和するために、JSON が利用している基盤の Dynamic type では、max_dynamic_paths パラメータを提供しており、別個のカラムファイルとして保存される一意のパス数を制限できます。このしきい値に達すると、それ以降のパスは共有のカラムファイルにコンパクトなエンコード形式で保存され、柔軟なデータのインジェストをサポートしつつ、パフォーマンスとストレージ効率を維持します。ただし、この共有カラムファイルへのアクセスは、専用カラムほど高パフォーマンスではありません。なお、JSON カラムは type hints と併用することもできます。"Hinted" カラムは専用カラムと同等のパフォーマンスを提供します。
• パスと型のより簡単な確認：JSON 型でも、推論された型やパスを確認するためのイントロスペクション関数が利用できますが、DESCRIBE などを使える静的な構造のほうが、より簡単に調査できる場合があります。

あるいは、1つのJSONカラムを持つテーブルを作成することもできます。

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`)

注記

JSON定義内のhost.name列とtimestamp列に型ヒントを指定しています。これらの列を順序付けキー/プライマリキーで使用するためです。これにより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、および elastic dump を同一のアベイラビリティゾーンまたはデータセンター内で実行することを推奨します。これにより、ネットワーク送信を最小化し、スループットを最大化できます。

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

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

データのストリーミング

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

# URLと認証情報をエクスポート
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

# 実行するコマンド - 必要に応じて変更
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 フラグ。
• point in time 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スキーマを参照）、同じinsertコマンドを使用できます。ただし、フォーマットとしてJSONEachRowではなく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で対応していることを前提としています。多くの場合、ユーザーはClickHouseへ挿入する前にElasticsearchデータのフィルタリングと変換を行う必要があります。

これはinputテーブル関数を使用することで実現できます。この関数を使用すると、標準出力に対して任意のSELECTクエリを実行できます。

先ほどのデータからtimestampとhostnameフィールドのみを保存する場合を想定します。ClickHouseスキーマは次のようになります：

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

elasticdumpからこのテーブルへデータを挿入するには、inputテーブル関数を使用します。JSON型により必要なカラムを動的に検出・選択できます。なお、この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 入力形式の使用が必要です。