@ chkit/ 插件回填

在 chkit 中检查 SQL 回填操作的 Beta 插件 。 它的API,配置,和CLI旗在发行之间可能会有变化.

部分 齐克特 单相机 此插件扩展 chkit 有数据回填命令的 CLI.

安装

bun add -d @chkit/plugin-backfill

使用量

在配置中注册插件 :

// clickhouse.config.ts
import { defineConfig } from '@chkit/core'
import { backfill } from '@chkit/plugin-backfill'

export default defineConfig({
  schema: './src/db/schema/**/*.ts',
  outDir: './chkit',
  plugins: [
    backfill(),
  ],
  clickhouse: {
    url: process.env.CLICKHOUSE_URL ?? 'http://localhost:8123',
  },
})

文档

见 回填插件文档 (β)。 。 。 .

SDK 软件

包根仅限于插件注册 API. CLI本身使用的一切——块规划器, SQL 构建器, Async 执行器, 记录器—— 也从 @chkit/plugin-backfill/sdk 子路径, 这样您可以建立自己的回填脚本而不通过 CLI .

import {
  generateChunkPlan,
  buildChunkExecutionSql,
  executeBackfill,
  getBackfillLogger,
  type ChunkPlan,
  type PlannerQuery,
} from '@chkit/plugin-backfill/sdk'

管道有三个阶段, 你可以使用任何子集:

1. 计划 —— 说 generateChunkPlan(...) 内览表格并返回 a ChunkPlan 描述如何将工作分割成大致相等的块.
2. 构建 SQL —— 说 buildChunkExecutionSql(...) 单曲 Chunk 变成 INSERT … SELECT 语句.
3. 执行 —— 说 executeBackfill(...) 以实数提交块 ClickHouseExecutor 有决定性的查询 身份证,投票,并恢复支持.

计划回填

generateChunkPlan 与任何 ClickHouse 客户端脱钩。 你通过一个 query 函数与 PlannerQuery 形状和规划者将它用于每个内向 / 探测 / 分解查询。 这使得策划者对仪器来说无足轻重,或者对另类客户奔跑.

import { createClient } from '@clickhouse/client'
import { generateChunkPlan, type PlannerQuery } from '@chkit/plugin-backfill/sdk'

const client = createClient({ url: process.env.CLICKHOUSE_URL })

const query: PlannerQuery = async (sql, settings) => {
  const result = await client.query({
    query: sql,
    format: 'JSONEachRow',
    clickhouse_settings: settings as Record<string, string | number | boolean>,
  })
  return result.json()
}

const plan = await generateChunkPlan({
  database: 'analytics',
  table: 'events',
  from: '2025-01-01T00:00:00Z',
  to: '2025-02-01T00:00:00Z',
  targetChunkBytes: 1_000_000_000, // ~1 GiB per chunk
  query,
  // 'count' is exact but slower; 'explain-estimate' is faster but approximate
  rowProbeStrategy: 'count',
})

console.log(`${plan.chunks.length} chunks, ${plan.totalRows.toLocaleString()} rows`)

对目标执行块

buildChunkExecutionSql 生成每圆 INSERT … SELECT 财务报告和审定财务报表 executeBackfill 以货币、投票和回调方式运行。 坚持下去 progress 用于支持恢复的参数.

import { createClickHouseExecutor } from '@chkit/clickhouse'
import {
  buildChunkExecutionSql,
  executeBackfill,
  type BackfillProgress,
} from '@chkit/plugin-backfill/sdk'

const executor = createClickHouseExecutor({
  url: process.env.CLICKHOUSE_URL!,
  username: 'default',
  password: process.env.CLICKHOUSE_PASSWORD!,
  database: 'analytics',
})

const chunksById = new Map(plan.chunks.map((chunk) => [chunk.id, chunk]))
let saved: BackfillProgress | undefined // load from disk for resume

const result = await executeBackfill({
  executor,
  planId: plan.planId,
  chunks: plan.chunks,
  buildQuery: ({ id }) =>
    buildChunkExecutionSql({
      planId: plan.planId,
      chunk: chunksById.get(id)!,
      target: 'analytics.events_backfill',
      table: plan.table,
    }),
  concurrency: 4,
  pollIntervalMs: 5_000,
  resumeFrom: saved,
  onProgress: async (progress) => {
    saved = progress
    // persist to disk / state store
  },
})

console.log(`done=${result.completed} failed=${result.failed}`)

计划的持久性

计划中包含可能包含非UTF-8字节的字符串边界(计划员使用) latin1- 字符串排序键的编码字节范围),所以 JSON 序列化 a ChunkPlan 直接丢失信息 。 当您需要通过存储来绕行一个计划时使用代码帮助器 :

import {
  encodeChunkPlanForPersistence,
  decodeChunkPlanFromPersistence,
} from '@chkit/plugin-backfill/sdk'

const json = JSON.stringify(encodeChunkPlanForPersistence(plan))
// later …
const plan2 = decodeChunkPlanFromPersistence(JSON.parse(json))

日志

计划员通过结构化的日志 @logtape/logtape 下划线 ['chkit', 'backfill'] 类别。 在进程开始时配置一个水槽以看到它们 —— 缓冲警告( > 5 s) 发布于 warning 职等,规划进展 info,并按战略作出决定 debug。 。 。 .

import { configureSync, getConsoleSink, getTextFormatter } from '@chkit/plugin-backfill/sdk'

configureSync({
  sinks: { console: getConsoleSink({ formatter: getTextFormatter({ timestamp: 'time' }) }) },
  loggers: [{ category: 'chkit', sinks: ['console'], lowestLevel: 'info' }],
  reset: true,
})

为了捕捉每个 SQL 语句, 计划员运行( 有时间、 服务器侧数据、 和每个策略分类), 包入您的 query 函数而不是仅仅依赖日志 —— 包装器在每次调用时都会看到原始的 SQL 和设置,并且可以在结构化日志的同时记录查询ID,响应头和持续时间.

许可证

麻省理工学院