把观测数据放进 ClickHouse 意味着什么 — schema、rollup、TTL 与职责划分

SELECT
    table,
    formatReadableSize(sum(data_uncompressed_bytes)) AS raw,
    formatReadableSize(sum(data_compressed_bytes))   AS compressed,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 1) AS ratio
FROM system.columns
WHERE database = 'otel'
GROUP BY table
ORDER BY sum(data_compressed_bytes) DESC;

-- 也可以按字段拆开看。压缩不下去的字段往往占了成本的大头
SELECT
    name,
    type,
    formatReadableSize(data_compressed_bytes) AS compressed,
    round(data_uncompressed_bytes / data_compressed_bytes, 1) AS ratio
FROM system.columns
WHERE database = 'otel' AND table = 'otel_traces'
ORDER BY data_compressed_bytes DESC
LIMIT 15;

CREATE TABLE otel.otel_traces
(
    Timestamp          DateTime64(9) CODEC(Delta(8), ZSTD(1)),
    TraceId            String CODEC(ZSTD(1)),
    SpanId             String CODEC(ZSTD(1)),
    ParentSpanId       String CODEC(ZSTD(1)),
    TraceState         String CODEC(ZSTD(1)),
    SpanName           LowCardinality(String) CODEC(ZSTD(1)),
    SpanKind           LowCardinality(String) CODEC(ZSTD(1)),
    ServiceName        LowCardinality(String) CODEC(ZSTD(1)),
    ResourceAttributes Map(LowCardinality(String), String) CODEC(ZSTD(1)),
    ScopeName          String CODEC(ZSTD(1)),
    ScopeVersion       String CODEC(ZSTD(1)),
    SpanAttributes     Map(LowCardinality(String), String) CODEC(ZSTD(1)),
    Duration           UInt64 CODEC(ZSTD(1)),
    StatusCode         LowCardinality(String) CODEC(ZSTD(1)),
    StatusMessage      String CODEC(ZSTD(1)),

    Events Nested (
        Timestamp  DateTime64(9),
        Name       LowCardinality(String),
        Attributes Map(LowCardinality(String), String)
    ) CODEC(ZSTD(1)),

    Links Nested (
        TraceId    String,
        SpanId     String,
        TraceState String,
        Attributes Map(LowCardinality(String), String)
    ) CODEC(ZSTD(1)),

    INDEX idx_trace_id       TraceId TYPE bloom_filter(0.001) GRANULARITY 1,
    INDEX idx_res_attr_key   mapKeys(ResourceAttributes)   TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_span_attr_key  mapKeys(SpanAttributes)       TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_duration       Duration TYPE minmax GRANULARITY 1
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (ServiceName, SpanName, toDateTime(Timestamp))
TTL toDateTime(Timestamp) + toIntervalDay(30)
SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;

CREATE TABLE otel.otel_logs
(
    Timestamp          DateTime64(9) CODEC(Delta(8), ZSTD(1)),
    TraceId            String CODEC(ZSTD(1)),
    SpanId             String CODEC(ZSTD(1)),
    TraceFlags         UInt8,
    SeverityText       LowCardinality(String) CODEC(ZSTD(1)),
    SeverityNumber     UInt8,
    ServiceName        LowCardinality(String) CODEC(ZSTD(1)),
    Body               String CODEC(ZSTD(1)),
    ResourceAttributes Map(LowCardinality(String), String) CODEC(ZSTD(1)),
    LogAttributes      Map(LowCardinality(String), String) CODEC(ZSTD(1)),

    -- 把经常用来过滤的键提升为物理字段
    HttpRoute    LowCardinality(String) MATERIALIZED LogAttributes['http.route'],
    HttpStatus   UInt16                 MATERIALIZED toUInt16OrZero(LogAttributes['http.response.status_code']),
    ErrorType    LowCardinality(String) MATERIALIZED LogAttributes['error.type'],

    INDEX idx_trace_id  TraceId    TYPE bloom_filter(0.001) GRANULARITY 1,
    INDEX idx_body      Body       TYPE tokenbf_v1(32768, 3, 0) GRANULARITY 1,
    INDEX idx_severity  SeverityNumber TYPE set(16) GRANULARITY 4
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (ServiceName, toStartOfFiveMinutes(Timestamp), Timestamp)
TTL toDateTime(Timestamp) + toIntervalDay(30)
SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;

-- JSON 类型的用法示例 — 给子字段数量设一个上限
CREATE TABLE otel.otel_logs_json
(
    Timestamp    DateTime64(9) CODEC(Delta(8), ZSTD(1)),
    ServiceName  LowCardinality(String) CODEC(ZSTD(1)),
    SeverityText LowCardinality(String) CODEC(ZSTD(1)),
    TraceId      String CODEC(ZSTD(1)),
    Body         String CODEC(ZSTD(1)),
    Attributes   JSON(max_dynamic_paths = 512) CODEC(ZSTD(1))
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (ServiceName, toStartOfFiveMinutes(Timestamp), Timestamp)
TTL toDateTime(Timestamp) + toIntervalDay(30)
SETTINGS ttl_only_drop_parts = 1;

-- 显式指定类型来访问子路径，这样才能用上索引和统计信息
SELECT
    ServiceName,
    Attributes.http.route::LowCardinality(String) AS route,
    count() AS c,
    quantile(0.99)(Attributes.duration_ms::Float64) AS p99
FROM otel.otel_logs_json
WHERE Timestamp >= now() - INTERVAL 1 HOUR
  AND SeverityText = 'ERROR'
GROUP BY ServiceName, route
ORDER BY c DESC
LIMIT 20;

-- 定义存储策略 (写在 config.xml 或单独的配置文件里)
-- hot: 本地 NVMe，cold: 对象存储

ALTER TABLE otel.otel_traces
MODIFY TTL
    toDateTime(Timestamp) + INTERVAL 3 DAY  TO VOLUME 'hot',
    toDateTime(Timestamp) + INTERVAL 14 DAY TO VOLUME 'cold',
    toDateTime(Timestamp) + INTERVAL 90 DAY DELETE;

-- 每个分区的大小与最早的数据
SELECT
    table,
    partition,
    formatReadableSize(sum(bytes_on_disk)) AS size,
    sum(rows) AS rows,
    min(min_time) AS oldest
FROM system.parts
WHERE database = 'otel' AND active
GROUP BY table, partition
ORDER BY partition
LIMIT 10;

-- 正在等待和正在进行的合并
SELECT table, elapsed, progress, num_parts, formatReadableSize(memory_usage) AS mem
FROM system.merges
WHERE database = 'otel';

-- part 数一多，插入就会开始被拒绝
SELECT table, count() AS parts
FROM system.parts
WHERE database = 'otel' AND active
GROUP BY table
ORDER BY parts DESC;

-- 1) 用来存放聚合结果的表
CREATE TABLE otel.trace_rollup_1m
(
    Bucket      DateTime,
    ServiceName LowCardinality(String),
    SpanName    LowCardinality(String),
    SpanKind    LowCardinality(String),
    Calls       AggregateFunction(count),
    Errors      AggregateFunction(countIf, UInt8),
    DurationQ   AggregateFunction(quantiles(0.5, 0.9, 0.99), Float64),
    DurationSum AggregateFunction(sum, Float64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(Bucket)
ORDER BY (ServiceName, SpanName, SpanKind, Bucket)
TTL Bucket + toIntervalDay(400);

-- 2) 原始表每次插入时自动做聚合
CREATE MATERIALIZED VIEW otel.trace_rollup_1m_mv TO otel.trace_rollup_1m AS
SELECT
    toStartOfMinute(Timestamp) AS Bucket,
    ServiceName,
    SpanName,
    SpanKind,
    countState()                                   AS Calls,
    countIfState(StatusCode = 'Error')             AS Errors,
    quantilesState(0.5, 0.9, 0.99)(Duration / 1e6) AS DurationQ,
    sumState(Duration / 1e6)                       AS DurationSum
FROM otel.otel_traces
GROUP BY Bucket, ServiceName, SpanName, SpanKind;

-- 3) 查询时把状态合并起来
SELECT
    ServiceName,
    SpanName,
    countMerge(Calls)                          AS calls,
    countIfMerge(Errors)                       AS errors,
    round(countIfMerge(Errors) / countMerge(Calls), 4) AS error_ratio,
    arrayElement(quantilesMerge(0.5, 0.9, 0.99)(DurationQ), 3) AS p99_ms
FROM otel.trace_rollup_1m
WHERE Bucket >= now() - INTERVAL 30 DAY
GROUP BY ServiceName, SpanName
ORDER BY calls DESC
LIMIT 30;

# otel-collector.yaml
exporters:
  clickhouse:
    endpoint: tcp://clickhouse.observability.svc:9000?dial_timeout=10s
    database: otel
    username: otel_writer
    password: ${env:CLICKHOUSE_PASSWORD}
    # 生产环境自己管理 DDL
    create_schema: false
    logs_table_name: otel_logs
    traces_table_name: otel_traces
    compress: lz4
    async_insert: true
    timeout: 10s
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_elapsed_time: 300s

processors:
  # ClickHouse 偏爱大批量。小批量插入太频繁会导致 part 暴增
  batch:
    timeout: 10s
    send_batch_size: 20000
    send_batch_max_size: 50000

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [clickhouse]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [clickhouse]

EXPLAIN indexes = 1
SELECT count()
FROM otel.otel_traces
WHERE ServiceName = 'checkout-api'
  AND Timestamp >= now() - INTERVAL 1 HOUR;
-- 读取到的 mark 数应该只占总数的极小一部分

processors:
  transform/normalize:
    error_mode: ignore
    trace_statements:
      - context: resource
        statements:
          # 把用旧名字进来的属性统一成当前的命名约定
          - set(attributes["deployment.environment.name"], attributes["deployment.environment"])
            where attributes["deployment.environment.name"] == nil
                  and attributes["deployment.environment"] != nil
          - delete_key(attributes, "deployment.environment")