chkit

Manage ClickHouse schemas and sync API data.

chkit is an open-source CLI for ClickHouse. Review migration SQL before applying it. Keep table definitions and API readers in your repository, alongside the code that uses them. Run the CLI from the terminal or CI.

TypeScript: schemas, migrations, and ingestion. Python: schemas and migrations through chkit-py.

Get started · Build a data source · Documentation

Beta: the public API is still evolving. Keep the CLI, core, and plugins on matching versions.

Why chkit

• Define tables and views. Keep ClickHouse tables, views, materialized views, and dictionaries in code. Import the schema from an existing database.
• Migrate and backfill data. Handle complex schema changes and backfill data across materialized views. Track progress and resume interrupted runs.
• Sync data from any source. Build reliable syncs from HTTP APIs, databases, and other sources into ClickHouse.
• Store raw or mapped records. Retain raw objects for SQL transformations, or map records into established entity tables before loading.
• Check schema drift in CI. Detect pending migrations, checksum mismatches, and live schema differences with chkit check.
• Generate types and backfill data. Plugins generate TypeScript types and Zod schemas or run SQL backfills. Agent skills guide schema and source authoring.

From a schema to a working sync

Start with a table, load 100 demo API records, then query posts per author. Change the view later using the data already stored in ClickHouse. This walkthrough uses TypeScript and the ingestion plugin. For a project using only chkit and @chkit/core, follow the schema tutorial.

bun add -d chkit@beta @chkit/core@beta @chkit/plugin-ingest@beta

Project configuration

Create clickhouse.config.ts and set the connection environment variables for the intended development database. Export the definitions below from src/chkit.ts.

import { defineConfig } from '@chkit/core'
import { ingest } from '@chkit/plugin-ingest'

export default defineConfig({
  entry: './src/chkit.ts',
  plugins: [ingest()],
  clickhouse: {
    url: process.env.CLICKHOUSE_URL ?? 'http://localhost:8123',
    username: process.env.CLICKHOUSE_USER ?? 'default',
    password: process.env.CLICKHOUSE_PASSWORD ?? '',
    database: 'default',
  },
})

The configuration guide covers existing projects and additional plugins.

1. Define a table

Create src/chkit.ts. Explicit columns describe the query shape; ingestionColumns adds the metadata used by the reader in step 3.

import { table } from '@chkit/core'
import { ingestionColumns } from '@chkit/plugin-ingest'

export const posts = table({
  database: 'default', name: 'posts',
  columns: [
    { name: 'id', type: 'String' },
    { name: 'title', type: 'String' },
    { name: 'user_id', type: 'UInt64' },
    ...ingestionColumns,
  ],
  engine: 'ReplacingMergeTree(_chkit_ingested_at)',
  primaryKey: ['id'], orderBy: ['id'],
})

2. Review and apply a migration

bunx chkit generate --name create-posts
bunx chkit migrate

# After reviewing the generated SQL and preview:
bunx chkit migrate --apply

3. Add a source reader

Add this to src/chkit.ts. The public demonstration API returns a bounded dataset, so this reader performs a full sync and maps each record into the table.

import { definePipeline, defineStream, HttpError } from '@chkit/plugin-ingest'

type Post = { id: number; title: string; userId: number }

const postStream = defineStream({
  id: 'demo.posts', destination: posts,
  async *read(context) {
    const page = await context.attempt(async (signal) => {
      const response = await fetch('https://jsonplaceholder.typicode.com/posts', { signal })
      if (!response.ok) throw await HttpError.fromResponse(response)
      return await response.json() as Post[]
    })
    yield { rows: page.map((post) => ({
      id: String(post.id), title: post.title, user_id: post.userId,
    })) }
  },
})

export const content = definePipeline({ id: 'content', streams: [postStream] })

bunx chkit ingest run --tag pipeline:content

The default loader writes the rows. For production sources, add pagination and incremental reads when the provider supports them.

4. Query through a view

Add this view to the same entry. FINAL reconciles repeated object versions before aggregation.

import { view } from '@chkit/core'

export const postsByAuthor = view({
  database: 'default', name: 'posts_by_author',
  as: `SELECT user_id, count() AS posts
       FROM default.posts FINAL
       GROUP BY user_id`,
})

bunx chkit generate --name posts-by-author
bunx chkit migrate
bunx chkit migrate --apply
bunx chkit query "SELECT * FROM default.posts_by_author ORDER BY user_id LIMIT 3"

Expected result for the demo dataset:

user_id	posts
1	10
2	10
3	10

If the final shape may change, retain raw records and transform in ClickHouse instead of mapping every field up front.

5. Evolve the model

Replace the previous view definition to add a measure:

import { view } from '@chkit/core'

export const postsByAuthor = view({
  database: 'default', name: 'posts_by_author',
  as: `SELECT user_id, count() AS posts,
         countIf(positionCaseInsensitive(title, 'qui') > 0) AS matching_posts
       FROM default.posts FINAL
       GROUP BY user_id`,
})

Generate, preview, and apply another migration. chkit recreates the ordinary view with the new SQL. The matching_posts measure uses already-stored rows: no API re-fetch or table backfill for this change.

6. Verify and repeat

bunx chkit check
bunx chkit ingest list
bunx chkit ingest run --tag pipeline:content

Use check in CI for migration state and schema drift. Schedule ingestion through cron, CI, or another job runner, with one ingestion process per target. Incremental sources resume from committed state; this full-sync demo reads the dataset again. See scheduling and recovery.

Set up chkit for your project

Goal	Start here
Manage a new schema	Getting started
Adopt an existing database	Pull a live schema
Implement an API source	Ingestion quickstart
Generate application types	TypeScript codegen
Recompute stored data	SQL backfills
Work with a coding agent	Agent skills

See the CLI reference for commands, flags, and JSON output.

Packages

Package	Description
chkit	CLI binary and command implementations
@chkit/core	Schema DSL, config, and diff engine
@chkit/clickhouse	ClickHouse client wrapper
@chkit/codegen	TypeScript type generation engine
@chkit/plugin-pull	Pull live schema into local files
@chkit/plugin-codegen	Codegen plugin for the CLI
@chkit/plugin-backfill	Backfill plugin for data migrations
@chkit/plugin-ingest	API source readers, batching, retries, and journaled checkpoints
@chkit/plugin-obsessiondb	ObsessionDB integration: auto-rewrite Shared engines for ClickHouse targets

Python

Install chkit-py (pip install chkit-py) to define schemas and run migrations, drift detection, and CI checks with Python config and schema files. API ingestion requires TypeScript. The Python source is in chkit_python/.

Documentation

Read the documentation at chkit.obsessiondb.com.

ObsessionDB

The ObsessionDB team builds chkit and provides a managed ClickHouse service:

• Engine configuration. Use the @chkit/plugin-obsessiondb plugin to select SharedReplacingMergeTree / SharedMergeTree for managed replication while keeping the same TypeScript schema for your development database.
• Managed infrastructure. ObsessionDB manages Keeper, replicas, and scaling.
• Release tests. The chkit release pipeline runs its E2E suite against ObsessionDB.

Try ObsessionDB →

Community

• @ObsessionDB on X: release notes and updates
• GitHub Issues: bugs and feature requests

Contributing

See CONTRIBUTING.md for development setup and guidelines.

License

MIT