# Mason-ETL

**Repository Path**: xmcqq/Mason-ETL

## Basic Information

- **Project Name**: Mason-ETL
- **Description**: ETL数据同步工具
- **Primary Language**: Go
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-04
- **Last Updated**: 2026-07-05

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 梅森 ETL (Mason-ETL)

高性能、分布式的异构数据库同步工具。支持 MySQL、PostgreSQL、Oracle、SQL Server、MongoDB、Elasticsearch 等十余种数据源之间的全量/增量/CDC 实时同步，内置可视化配置与调度管理界面，面向亿级数据量场景设计。

## 特性

- **高性能**: 多表并行 + 分片并行同步，支持亿级数据量
- **多数据源**: 11 种数据库支持（MySQL / PostgreSQL / Oracle / SQLServer / MongoDB / GaussDB / GBase / 人大金仓 / Elasticsearch / Hive）
- **异构互转**: 任意数据源之间的数据同步（如 MySQL → PostgreSQL）
- **丰富转换**: 字段映射、脱敏、字典翻译、时间格式化、计算字段、字符串处理
- **CDC 实时同步**: MySQL Binlog / PostgreSQL WAL / MongoDB Change Streams
- **分布式**: Server + Worker 架构，支持多节点并行执行
- **定时调度**: Cron 表达式 + 可视化配置
- **Web 管理**: Element Plus 管理界面，实时监控 + 进度 + 日志
- **断点续传**: 分片级检查点，支持失败重试

## 快速开始

### Docker 部署（推荐）

```bash
# 启动全部服务
docker-compose up -d

# 启动完整集群（含 2 个 Worker）
docker-compose --profile full up -d

# 访问
# Web UI:  http://localhost:3000
# API:     http://localhost:8080
# NATS监控: http://localhost:8222
```

### 本地开发

```bash
# 要求
# - Go 1.22+
# - Node.js 20+
# - PostgreSQL 15+

# 1. 启动 PostgreSQL 并创建数据库
createdb mason_etl

# 2. 配置 config/server.yaml 中的数据库连接信息

# 3. 启动后端
go run ./cmd/server/

# 4. 启动前端（新终端）
cd web && npm install && npm run dev

# 访问 http://localhost:3000
```

## 项目结构

```
Mason-ETL/
├── cmd/
│   ├── server/main.go      # 主节点入口
│   └── worker/main.go      # Worker 节点入口
├── internal/
│   ├── api/                # HTTP API (Gin Router + Handler)
│   ├── cdc/                # CDC 引擎 (Binlog/WAL/ChangeStreams)
│   ├── config/             # 配置管理 (Viper)
│   ├── connection/         # 连接管理 (Driver 注册中心 + 连接池)
│   ├── engine/             # 执行引擎 (Pipeline + 分片 + 进度)
│   ├── model/              # 数据模型 (7 个表)
│   ├── plugin/             # 插件接口 (Reader/Writer)
│   ├── reader/             # 读取器 (11 种数据源)
│   ├── repository/         # 数据访问层
│   ├── scheduler/          # 调度器 (Cron + Dispatcher + 队列)
│   ├── service/            # 业务服务 (进度推送 + 监控)
│   ├── transform/          # 转换管道 (9 种转换器)
│   ├── worker/             # Worker 节点 (执行器 + 心跳)
│   └── writer/             # 写入器 (11 种目标)
├── pkg/                    # 可复用工具包
│   ├── checkpoint/         # 断点续传
│   ├── crypto/             # AES-256-GCM 加密
│   ├── logger/             # Zap 日志封装
│   ├── retry/              # 指数退避重试
│   └── types/              # 通用类型
├── web/                    # Vue 3 前端
│   ├── src/
│   │   ├── api/            # API 封装 (Axios)
│   │   ├── components/     # 公共组件
│   │   ├── router/         # 路由配置
│   │   ├── store/          # Pinia 状态
│   │   └── views/          # 页面组件
│   └── package.json
├── config/                 # 配置模板
├── docs/                   # 文档
├── docker-compose.yaml     # Docker 编排
├── Dockerfile.server       # Server 镜像
├── Dockerfile.worker       # Worker 镜像
├── Dockerfile.web          # Web 镜像
├── Makefile                # 构建脚本
└── README.md
```

## 数据源支持

| 数据源 | 读取 | 写入 | CDC | 状态 |
|--------|------|------|-----|------|
| MySQL | ✅ | ✅ | ✅ Binlog | 生产就绪 |
| PostgreSQL | ✅ | ✅ | ✅ WAL | 生产就绪 |
| SQL Server | ✅ | ✅ | 🔧 | 完整实现 |
| MongoDB | ✅ | ✅ | ✅ Changes | 生产就绪 |
| Elasticsearch | ✅ | ✅ | 🔧 | 完整实现 |
| GaussDB | ✅ | ✅ | - | PG 兼容模式 |
| GBase | ✅ | ✅ | - | MySQL 兼容模式 |
| 人大金仓 | ✅ | ✅ | - | PG 兼容模式 |
| Oracle | 🔧 | 🔧 | 🔧 | 需安装 godror |
| Hive | 🔧 | 🔧 | - | 需安装 gohive |

## 同步模式

| 模式 | 说明 | 适用场景 |
|------|------|---------|
| `full` | 全量同步 | 首次数据迁移、全量刷新 |
| `incremental` | 增量同步 | 定时抓取新增数据 |
| `cdc` | CDC 实时同步 | 数据库变更实时捕获 |

## 转换规则

| 规则类型 | 说明 | 示例 |
|---------|------|------|
| `field_mapping` | 字段映射 | `user_id` → `id` |
| `field_desensitize` | 数据脱敏 | 手机号中间4位掩码 |
| `dict_translate` | 字典翻译 | `0` → `无效`, `1` → `有效` |
| `datetime_format` | 时间格式化 | `yyyy-MM-dd` → `2006-01-02` |
| `string_op` | 字符串操作 | trim / upper / lower / replace |
| `computed_field` | 计算字段 | `concat(first_name, '-', last_name)` |
| `type_convert` | 类型转换 | int → varchar |
| `field_add` | 添加字段 | 固定值 / 默认值 |
| `field_remove` | 删除字段 | 移除敏感列 |

## API 文档

```
GET    /api/v1/health                         健康检查
GET    /api/v1/datasources                    数据源列表
POST   /api/v1/datasources                    创建数据源
POST   /api/v1/datasources/:id/test           测试连接
GET    /api/v1/tasks                          任务列表
POST   /api/v1/tasks                          创建任务
POST   /api/v1/tasks/:id/publish              发布任务
POST   /api/v1/jobs/run                       手动触发作业
GET    /api/v1/jobs/:id                       作业详情+进度
GET    /api/v1/schedules                      调度列表
POST   /api/v1/schedules                      创建调度
GET    /api/v1/logs                           日志查询
GET    /api/v1/monitor/overview               监控概览
WS     /api/v1/ws/progress/:job_id            实时进度推送
```

## 配置说明

主要配置项 (`config/server.yaml`):

```yaml
server:
  port: 8080           # API 端口
database:
  host: 127.0.0.1      # 元数据库地址
  port: 5432
  user: mason
  password: mason123
  dbname: mason_etl
nats:
  enabled: false       # 启用分布式模式时设为 true
redis:
  enabled: false       # 可选，用于缓存
worker:
  max_jobs: 4          # 每个 Worker 最大并发作业数
```

## License

MIT