Cloud or Self-Hosted

ClickHouse 适配器

HTTP 接口异步插入

通过 HTTP 接口将宽事件写入 ClickHouse——对需要聚合的内容使用类型化列，其余内容完整保留为 JSON。

ClickHouse 适配器通过HTTP 接口，以 JSONEachRow 格式写入宽事件。它支持本地实例、自管理集群以及 ClickHouse Cloud。

默认架构会为你进行筛选和聚合的字段提供类型化列，并将完整的宽事件以 JSON 形式存储在 data 中，因此不会丢失任何字段，而且向事件中添加字段也永远不需要迁移。

添加 ClickHouse drain 适配器

安装

ClickHouse 适配器已随 evlog 一起提供：

src/index.ts

import { createClickHouseDrain } from 'evlog/clickhouse'

创建表

在接入排空程序之前运行一次：

schema.sql

CREATE TABLE IF NOT EXISTS evlog_events
(
  timestamp     DateTime64(3, 'UTC'),
  level         LowCardinality(String),
  service       LowCardinality(String),
  environment   LowCardinality(String),
  request_id    String,
  trace_id      String,
  span_id       String,
  method        LowCardinality(String),
  path          String,
  status        Nullable(UInt16),
  duration      String,
  duration_ms   Nullable(UInt32),
  error_name    String,
  error_message String,
  data          String
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(timestamp)
ORDER BY (service, environment, timestamp)
TTL toDateTime(timestamp) + INTERVAL 30 DAY;

ORDER BY (service, environment, timestamp) 与你最常用的筛选方式相匹配。根据你的保留策略调整 TTL，如果要永久保留事件，则完全删除该子句。

快速开始

// server/plugins/evlog.ts
import { createClickHouseDrain } from 'evlog/clickhouse'

export default defineNitroPlugin((nitroApp) => {
  nitroApp.hooks.hook('evlog:drain', createClickHouseDrain())
})

import { Hono } from 'hono'
import { evlog } from 'evlog/hono'
import { createClickHouseDrain } from 'evlog/clickhouse'

const app = new Hono()
app.use(evlog({ drain: createClickHouseDrain() }))

import express from 'express'
import { evlog } from 'evlog/express'
import { createClickHouseDrain } from 'evlog/clickhouse'

const app = express()
app.use(evlog({ drain: createClickHouseDrain() }))

import Fastify from 'fastify'
import { evlog } from 'evlog/fastify'
import { createClickHouseDrain } from 'evlog/clickhouse'

const app = Fastify()
await app.register(evlog, { drain: createClickHouseDrain() })

import { Elysia } from 'elysia'
import { evlog } from 'evlog/elysia'
import { createClickHouseDrain } from 'evlog/clickhouse'

const app = new Elysia().use(evlog({ drain: createClickHouseDrain() }))

import { Module } from '@nestjs/common'
import { EvlogModule } from 'evlog/nestjs'
import { createClickHouseDrain } from 'evlog/clickhouse'

@Module({
  imports: [EvlogModule.forRoot({ drain: createClickHouseDrain() })],
})
export class AppModule {}

import { initLogger } from 'evlog'
import { createClickHouseDrain } from 'evlog/clickhouse'

initLogger({
  env: { service: 'my-app' },
  drain: createClickHouseDrain(),
})

配置

环境变量

变量	必填	描述
`CLICKHOUSE_ENDPOINT`	是	HTTP 接口 URL（也接受 `CLICKHOUSE_URL`）
`CLICKHOUSE_USER`	否	用户名。默认为 `default`
`CLICKHOUSE_PASSWORD`	否	密码。未经身份验证的本地实例可省略
`CLICKHOUSE_DATABASE`	否	数据库。默认为 `default`
`CLICKHOUSE_TABLE`	否	表。默认为 `evlog_events`

选项

选项	类型	默认值	描述
`endpoint`	`string`	—	HTTP 接口 URL
`database`	`string`	`default`	目标数据库
`table`	`string`	`evlog_events`	目标表
`username`	`string`	`default`	用户名
`password`	`string`	—	密码
`asyncInsert`	`boolean`	`true`	在服务器端批量插入
`waitForAsyncInsert`	`boolean`	`false`	响应前等待刷新完成
`transform`	`(event) => Record<string, unknown>`	`toClickHouseRow`	将事件映射为一行
`timeout`	`number`	`5000`	请求超时时间（毫秒）
`retries`	`number`	`2`	重试次数

部署

无论采用哪种方式，适配器都相同。只有端点和凭据会发生变化。

自托管

本地或自行管理的实例，可选择启用或不启用身份验证：

server/plugins/evlog.ts

createClickHouseDrain({ endpoint: 'http://localhost:8123' })

.env

CLICKHOUSE_ENDPOINT=http://localhost:8123
# 仅当你的实例需要身份验证时：
CLICKHOUSE_USER=evlog
CLICKHOUSE_PASSWORD=your-password

全新的容器以 default 用户运行且没有密码，这就是为什么 username 默认为 default，而 password 是可选的。

ClickHouse Cloud

云服务始终需要凭据，并通过 HTTPS 的 8443 端口进行通信：

server/plugins/evlog.ts

createClickHouseDrain({
  endpoint: 'https://abc123.eu-west-1.aws.clickhouse.cloud:8443',
  password: process.env.CLICKHOUSE_PASSWORD,
  database: 'logs',
})

.env

CLICKHOUSE_ENDPOINT=https://abc123.eu-west-1.aws.clickhouse.cloud:8443
CLICKHOUSE_PASSWORD=your-service-password
CLICKHOUSE_DATABASE=logs

从 ClickHouse Cloud 控制台的 Connect → HTTPS 中复制端点，包括 :8443 端口。云服务会在空闲时暂停，因此暂停后的第一次插入可能需要几秒钟；如果看到中止错误，请增大 timeout。

凭据通过 X-ClickHouse-User / X-ClickHouse-Key请求头发送，绝不会作为查询参数发送，因此不会出现在 system.query_log 或中间代理的访问日志中。

在本地验证

在 Docker 中启动 ClickHouse，创建表并推送一个真实事件：

Terminal

docker compose -f packages/evlog/test/e2e/docker-compose.yml up -d
CLICKHOUSE_ENDPOINT=http://localhost:8123 pnpm run test:e2e
docker compose -f packages/evlog/test/e2e/docker-compose.yml down -v

该测试套件是一个往返测试：它插入事件，使用 SELECT 将其读回，并断言类型化列和 data JSON 均得以保留。 compose 文件会在首次启动时根据上述架构创建 evlog_events。如果没有设置 CLICKHOUSE_ENDPOINT，测试会显示醒目标记并跳过自身。

若要查看事件而不是对其进行断言，请填充沙盒并打开已配置的仪表板。该技术栈附带一个已连接 ClickHouse 数据源的 Grafana：

Terminal

pnpm run sandbox:up
pnpm run sandbox:seed
# → http://localhost:3001/d/evlog-wide-events

ClickHouse 内置的 /play 页面 http://localhost:8123/play 也可以用于执行原始 SQL，但它有意保持简洁，而仪表板是更好的起点。

异步插入

日志摄取意味着会有许多小型插入操作，而每个请求对应一个 MergeTree 分区会很快导致表性能下降。因此，适配器默认启用异步插入，并且不会等待刷新完成：

async_insert=1&wait_for_async_insert=0

ClickHouse 会在服务器端缓冲行，并批量写入。排空操作不会因磁盘写入而阻塞请求。

如果需要在 drain 解析前确认插入已持久化，请设置 waitForAsyncInsert: true，代价是增加延迟。设置 asyncInsert: false 可进行同步插入，但这只有在你已经通过 evlog/pipeline 在客户端完成批处理时才有意义。

查询

按排序键对类型化列建立索引；其他所有内容都存储在 data 中，并可通过 ClickHouse 的 JSON 函数访问：

-- 过去一小时内每个服务的错误率
SELECT service, countIf(level = 'error') / count() AS error_rate
FROM evlog_events
WHERE timestamp > now() - INTERVAL 1 HOUR
GROUP BY service;

-- 最慢的路由
SELECT path, count() AS hits, avg(duration_ms) AS avg_ms, quantile(0.95)(duration_ms) AS p95_ms
FROM evlog_events
WHERE timestamp > now() - INTERVAL 1 DAY AND method = 'GET' AND duration_ms IS NOT NULL
GROUP BY path
ORDER BY avg_ms DESC
LIMIT 20;

-- 一个请求的端到端记录
SELECT timestamp, level, path, status, data
FROM evlog_events
WHERE request_id = '4a8ff3a8-...'
ORDER BY timestamp;

-- 访问仅保存在 data 中的自定义字段
SELECT JSONExtractString(data, 'user', 'plan') AS plan, count()
FROM evlog_events
WHERE timestamp > now() - INTERVAL 7 DAY
GROUP BY plan;

自定义架构

传入 transform 以针对自定义结构的表：

server/plugins/evlog.ts

import { createClickHouseDrain } from 'evlog/clickhouse'

createClickHouseDrain({
  table: 'app_logs',
  transform: event => ({
    ts: event.timestamp,
    lvl: event.level,
    svc: event.service,
    payload: JSON.stringify(event),
  }),
})

键必须与列名匹配，因为 ClickHouse 会拒绝包含未知列的 JSONEachRow 插入。

批处理

与管道配合使用，以更大的批次插入：

server/plugins/evlog.ts

import { createDrainPipeline } from 'evlog/pipeline'
import { createClickHouseDrain } from 'evlog/clickhouse'
import type { DrainContext } from 'evlog'

const pipeline = createDrainPipeline<DrainContext>({
  batch: { size: 500, intervalMs: 5000 },
})

export default defineNitroPlugin((nitroApp) => {
  nitroApp.hooks.hook('evlog:drain', pipeline(createClickHouseDrain()))
})

故障排除

Missing endpoint。 未设置 CLICKHOUSE_ENDPOINT，且未传入 endpoint。排空程序会记录一次 [evlog/clickhouse] Missing endpoint，之后不再执行任何操作，因此请求永远不会被阻塞或失败。

ClickHouse API error: 401。 检查 CLICKHOUSE_USER / CLICKHOUSE_PASSWORD。凭据通过 X-ClickHouse-User / X-ClickHouse-Key 请求头发送，绝不会出现在查询字符串中，因此不会进入 system.query_log。

提及未知列的 ClickHouse API error: 400。 你的表结构与行数据形状不匹配。请根据上面的 DDL 创建表，或提供与列匹配的 transform。

Cannot parse input。 表已存在，但某一列的类型不匹配。status 是 Nullable(UInt16)，因为非 HTTP 事件中没有该字段。

直接使用 API

scripts/backfill.ts

import { sendBatchToClickHouse, sendToClickHouse } from 'evlog/clickhouse'

await sendToClickHouse(event, { endpoint: 'http://localhost:8123' })
await sendBatchToClickHouse(events, { endpoint: 'http://localhost:8123' })

toClickHouseRow()、toJSONEachRow() 和 resolveClickHouseUrl() 也已导出。

后续步骤

Sampling：在数据到达 ClickHouse 之前控制数据量
Enrichers：为每个事件添加派生字段
Pipeline：批处理、重试和扇出
Adapters Overview：所有可用目标亚洲欧美

Edit this pageorReport an issue

Grafana Loki

将宽事件推送到 Grafana Loki——自托管、多租户或 Grafana Cloud——使用低基数标签并支持完整的 JSON 查询。

OTLP

通过 OpenTelemetry 协议（OTLP）向 Grafana、Datadog、Honeycomb 以及任何兼容的后端发送日志。支持 gRPC 和 HTTP 传输。