openetl-go

README

OpenETL-Go

Single-binary ETL/CDC orchestration for moving, shaping, and operating data pipelines.

中文 README

OpenETL-Go runs Source -> Transform -> Sink pipelines from one binary. It can stay simple for file or database sync jobs, then grow into DAG orchestration, parallel execution, and master-worker distributed processing when a pipeline needs multiple sources, branches, joins, windows, or sinks.

The project does not keep a separate wide-table product area. Denormalized detail tables and real-time aggregate tables are expressed as normal pipeline or DAG specs using sources, transforms, state, and sinks.

What It Does

Area	Capability
Pipeline orchestration	Linear pipelines, DAG nodes/edges, conditional routing, fanout, parallel shards, scheduled/streaming execution
Data movement	CDC, batch, stream, file, HTTP, Redis, object storage, warehouse and search/index sinks
Data shaping	Filter, validate, type conversion, rename/drop/add fields, envelope normalization, lookup/enrichment, join, tumbling windows, deduplicate, Lua/JS/TS/WASM extension points
Operations	Web UI, REST API, saved connection catalog, pipeline validation, connection test, transform dry-run, Prometheus metrics, audit log
Reliability	At-least-once delivery by default, checkpoints, retry/backoff, DLQ list/replay/delete, idempotent sink modes where supported
Runtime	SQLite standalone mode, MySQL/PostgreSQL shared storage, master-worker distributed dispatch

Connector coverage is broad, but maturity is not identical across every connector and edge case. Treat the default contract as at-least-once, then use business keys, versions, upserts, or sink-specific idempotency to remove duplicate effects. See idempotency and the roadmap for current production-readiness notes.

Quick Start

Run the bundled MySQL CDC to ClickHouse demo:

podman compose -f docker-compose.quickstart.yml up -d

Then open:

• Web UI and proxied API: http://localhost:8000
• Direct ETL API: http://localhost:8001

Example specs are loaded from pipes-quickstart/. The full walkthrough is in docs/quickstart.md.

Minimal Pipeline Spec

Pipeline specs are YAML files under pipes/ or the configured etl.specsDir.

name: file-to-file
source:
  type: file
  config:
    path: /app/data/input/orders.jsonl
    format: json
sink:
  type: file_sink
  config:
    output_dir: /app/data/output

Full connector fields are documented in docs/etl-config-schema.md.

Reusing Saved Connections

Connections created through the UI or POST /api/v2/connections can be used by linear pipelines and DAG nodes. The saved connection supplies kind, type, and shared config; inline config overrides per-pipeline fields such as table, topic, query, or output path.

source:
  connection: orders-mysql
  config:
    table: orders
sink:
  connection_ref: warehouse-clickhouse
  config:
    table: orders_wide

The DAG designer can select saved connections directly, and connection tests can run before a spec is saved or started.

Advanced Aggregation By Orchestration

Wide-table and real-time aggregate use cases are built from ordinary pipeline pieces:

Kafka/MySQL CDC facts
  -> normalize_envelope / filter / type_convert
  -> lookup or join dimension data
  -> optional deduplicate and tumbling window aggregate
  -> ClickHouse / MySQL / PostgreSQL / Doris / S3 / Kafka sink

Current examples:

• testdata/pipes-wide-table/kafka-orders-detail-clickhouse.yaml - Kafka order events enriched with MySQL dimensions and written to ClickHouse.
• testdata/pipes-wide-table/kafka-orders-aggregate-clickhouse.yaml - Kafka order events deduplicated, enriched, window-aggregated, and written to ClickHouse.
• docs/adr/0001-kafka-wide-table.zh.md - design note for the orchestration approach.

This means the pipeline engine has the building blocks for denormalized detail tables and tumbling-window aggregates today. More complex stream-stream joins, sliding/session windows, CDC dimension updates, late-data handling, and DAG/stateful replay still need tighter production certification; those gaps are tracked in the roadmap rather than in a separate module.

Connectors And Operators

Stage	Built-in surface
Sources	mysql_cdc, mysql_snapshot_cdc, postgres_cdc, mysql_batch, kafka, file, http, redis
Transforms	normalize_envelope, filter, validate, type_convert, rename, drop_field, add_field, deduplicate, lookup, enricher, join, window, router, fanout, tap, rate_limiter, lua, javascript, typescript, WASM plugins
Sinks	clickhouse, mysql, postgres/postgresql, doris, elasticsearch/es, kafka, redis, s3, file_sink, jdbc

For exact fields, defaults, secret markers, and examples, use the plugin schema API (GET /api/v2/plugins/schema) or docs/etl-config-schema.md.

Run And Build

Download a release archive from Releases, or run the container image:

docker run -d --name openetl-go -p 8000:8000 -p 8001:8001 \
  -v "$PWD/pipes:/app/pipes" \
  ghcr.io/a8851625/openetl-go:latest

Build from source:

make build

Useful development commands:

make test          # unit tests with -race
make test-quick    # faster internal ETL test loop

cd web
npm install
npm run build      # rebuild resource/public

Optional runtime builds:

Build option	Effect
default	Pure Go core plus built-in connectors and Lua
-tags=extism	Enable WASM plugin runtime
-tags=nolua	Remove Lua runtime for a smaller binary
CGO_ENABLED=1	Enable JavaScript/TypeScript transforms through QuickJS

Runtime Model

• Config: manifest/config/config.yaml.
• Specs: YAML files under pipes/ or etl.specsDir, hot-reloaded by file watch.
• Storage: SQLite by default; MySQL/PostgreSQL for shared state and distributed mode.
• API auth: set ETL_API_TOKEN, then use X-API-Token or Authorization: Bearer <token>.
• Metrics: Prometheus endpoint at /metrics.
• UI/API: GoFrame serves the Web UI on :8000 and proxies /api/v2/* to the ETL API server on :8001.

Documentation

• Quick start / 中文
• REST API / 中文
• YAML config reference / 中文
• Idempotency and delivery semantics / 中文
• Parallelism and batching / 中文
• Roadmap and maturity notes
• Architecture standard
• Contributing / 中文

License

MIT, see LICENSE.