# realtime-pipeline

**Repository Path**: grow-project/realtime-pipeline

## Basic Information

- **Project Name**: realtime-pipeline
- **Description**: go 实时 OLAP 数据分析管道
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-06-28
- **Last Updated**: 2026-06-28

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 实时 OLAP 数据分析管道

[![ci](https://github.com/liukai6/realtime-pipeline/actions/workflows/ci.yml/badge.svg)](https://github.com/liukai6/realtime-pipeline/actions/workflows/ci.yml)
![go](https://img.shields.io/badge/go-1.26-blue)
![coverage](https://img.shields.io/badge/coverage-71%25-yellow)

> **语言**：中文 · [English](./README.md)

一条端到端的实时数据分析管道，使用 Go 实现：HTTP / Kafka 入口、WAL 持久化、基于 DAG 的流式计算与 Worker Pool、面向 ClickHouse / Elasticsearch 的**幂等**批量写入、基于 Checkpoint 的**Exactly-Once** 故障恢复。

> 实施计划：[`PLAN.md`](./PLAN.md) ·
> 架构深入：[`docs/architecture.md`](./docs/architecture.md) ·
> 运维手册：[`docs/operations.md`](./docs/operations.md) ·
> 基准报告：[`docs/benchmark.md`](./docs/benchmark.md)

---

## 总览

| 能力 | 实现位置 | 验证手段 |
|---|---|---|
| HTTP 单条 / 批量入口 | `internal/source/http.go` | 单测 89.9% + 冒烟 |
| Kafka consumer group 入口 | `internal/source/kafka.go` | testcontainers 集成测试 |
| 三级自适应背压 | `internal/source/ratelimit.go` | 单测 100% |
| WAL 追加写 + CRC + 分段滚动 | `internal/fault/wal.go` | 单测 87.9%，Bench 183k EPS |
| DAG + Worker Pool + Fan-in/Fan-out | `internal/engine/*` | 单测 92.0%，Bench 498k EPS（5 节点） |
| 滚动 / 滑动 / 会话窗口 | `internal/engine/window.go` | 单测 + 窗口状态 checkpoint |
| 双门限批量写入 | `internal/writer/batch.go` | 单测 + 对单条写入 534× 加速 |
| ClickHouse `ReplacingMergeTree` 幂等 | `internal/writer/clickhouse_driver.go` | 集成测试：同批写 3 次 → 行数不变 |
| Elasticsearch `_id` 幂等 | `internal/writer/elasticsearch_driver.go` | 集成测试：同批 bulk 3 次 → 计数不变 |
| 自适应批次大小（× 1.2 / × 0.5） | `internal/writer/adaptive.go` | 单测 |
| DLQ 终态失败托底 | `internal/writer/dlq.go` | 单测 |
| Checkpoint + WAL 回放（Exactly-Once） | `internal/fault/checkpoint*` | 单测 + chaos kill -9 → 零丢失 |
| 8 类 Prometheus 指标 + Grafana 看板 | `internal/monitor/*` + `deployments/grafana/*` | 看板已入仓 |
| 合成压测工具 | `cmd/loadtest` | 实测 41k EPS @ M2 |

---

## 快速开始

```bash
# 1. 拉起第三方依赖（CH、ES、Kafka KRaft、Redis、Prometheus、Grafana）
make up

# 2. 运行管道
make run

# 3. 发几条事件
curl -s -X POST http://localhost:8083/v1/events/batch \
  -H 'Content-Type: application/json' \
  -d '{"events":[
        {"event_id":"a1","timestamp":"2026-06-28T00:00:00Z","source":"http","type":"click"},
        {"event_id":"a2","timestamp":"2026-06-28T00:00:00Z","source":"http","type":"view"}
      ]}'

# 4. 看监控
open http://localhost:3000     # Grafana, admin/admin
open http://localhost:9090     # Prometheus

# 5. 压测
go run ./cmd/loadtest -addr=http://localhost:8082 -qps=580 -burst-qps=5800 \
                      -burst-for=10s -duration=30s -batch=20 -parallelism=8
```

---

## 数据流总览

```
HTTP /v1/events  ─┐
                   ├─→ Source ─→ 自适应限流 ─→ WAL ─→ Engine (DAG) ─→ Batcher ─→ ClickHouse + ES
Kafka topic ──────┘                                                                       │
                                                                                          ▼
                                                                                       DLQ.jsonl

                                Checkpoint Manager：每 cfg.fault.checkpoint.interval 触发一次快照
                                                    → FileStore (atomic 写入) + Redis (镜像)
                                启动时：Restore 最新快照 → 从 cp.WALOffset+1 回放 WAL
```

完整架构图见 [`docs/architecture.md`](./docs/architecture.md)。

---

## HTTP 端点

| 端点 | 用途 |
|---|---|
| `POST /v1/events` (端口 8081) | 单条事件入口 |
| `POST /v1/events/batch` | 每请求最多 1000 条；部分成功返回 `207 Multi-Status` |
| `GET /healthz` (端口 8080) | 存活探针 |
| `GET /readyz` | 就绪探针 |
| `GET /metrics` | Prometheus 抓取端点 |
| `GET /debug/pprof/*` | 运行时性能分析（CPU / heap / goroutine / ...） |

---

## Prometheus 指标

| 名称 | 类型 | 标签 |
|---|---|---|
| `pipeline_source_events_total` | Counter | `source` |
| `pipeline_source_backpressure_rejected_total` | Counter | `source` |
| `pipeline_event_latency_ms` | Histogram | — |
| `pipeline_channel_buffer_usage` | Gauge | `name` |
| `pipeline_writer_batch_size` | Histogram | — |
| `pipeline_writer_write_errors_total` | Counter | `writer` |
| `pipeline_fault_checkpoint_duration_ms` | Histogram | — |
| `pipeline_fault_wal_sync_duration_ms` | Histogram | — |

涵盖以上指标的 8 条告警规则见
[`deployments/prometheus/alert.rules.yml`](./deployments/prometheus/alert.rules.yml)。

---

## 配置

`configs/pipeline.yaml` 列出了每一项配置。任何字段都可以通过环境变量
**`PIPELINE_<节>_<键>`** 覆盖，例如：

```bash
PIPELINE_SERVER_ADDR=":9090" \
PIPELINE_WRITER_MAX_BATCH_SIZE=20000 \
PIPELINE_FAULT_CHECKPOINT_INTERVAL=2s \
go run ./cmd/pipeline
```

DAG 拓扑定义在 [`configs/dag.yaml`](./configs/dag.yaml)，跨进程热替换。

---

## 开发命令

```bash
make help          # 查看所有目标
make fmt           # gofmt + goimports
make lint          # golangci-lint（15 个 linter，严格模式）
make test          # 单元测试
make test-race     # 加 -race
make cover         # 生成覆盖率 HTML
make bench         # 全部基准测试
make vuln          # govulncheck 漏洞扫描
make test-integration   # testcontainers 集成测试（需要 Docker）
make ci            # 完整本地 CI
```

---

## 目录结构

```
realtime-pipeline/
├── cmd/
│   ├── pipeline/      # 生产二进制入口
│   └── loadtest/      # 合成 HTTP 压测工具
├── configs/
│   ├── pipeline.yaml  # 默认管道配置
│   └── dag.yaml       # 默认 DAG 拓扑（5 节点 fan-out）
├── deployments/
│   ├── clickhouse/init.sql
│   ├── prometheus/{prometheus.yml,alert.rules.yml}
│   └── grafana/{provisioning,dashboards}
├── docs/
│   ├── architecture.md    # 架构详解
│   ├── benchmark.md       # 基准报告
│   └── operations.md      # 运维手册
├── internal/
│   ├── config/        # viper 配置加载
│   ├── engine/        # DAG + 算子 + 窗口
│   ├── fault/         # WAL + Checkpoint + Manager
│   ├── logger/        # zap 封装
│   ├── monitor/       # Prometheus 指标
│   ├── server/        # 管理 HTTP（health / metrics / pprof）
│   ├── source/        # HTTP + Kafka source + 自适应限流
│   └── writer/        # Batcher + CH + ES + DLQ
├── pkg/event/         # 公共 Event 类型
├── docker-compose.yml
├── Makefile
└── PLAN.md
```

---

## 许可与致谢

内部探索性项目，授权方式由作者酌定。所有第三方依赖遵循各自许可证。