SKILL: Addax 项目知识

1. 项目整体认识

1.1 项目定位

名称：Addax
类型：通用开源 ETL 工具（Extract–Transform–Load）
起源：基于阿里巴巴 DataX 的 fork 与演进
目标：在多种异构数据源之间，提供稳定、高效、可扩展的“离线数据同步”能力

1.2 核心价值

支持 20+ SQL/NoSQL/文件/时序/大数据 数据源
使用 JSON 任务配置 即可完成复杂同步，无需写代码
插件化架构，Reader / Writer / Transformer 解耦，可自由扩展
提供 数据质量监控、速率控制、错误容忍、脏数据探测 等生产级能力
既可命令行运行，也可通过 Server 模块 HTTP 接口 异步提交和管理任务
有配套的 addax-admin / addax-ui 项目做 Web 管控

2. 概念与架构模型

2.1 核心业务概念

在与用户讨论 / 理解需求时，应优先按以下抽象模型理解：

Job（作业）
- 一次完整的数据同步任务，从一个源到一个目标
- 通过一个 JSON 文件描述：数据源 reader、目标端 writer、变换规则、速率控制、错误阈值等
- Job 是业务上的最小单位，如 “从 MySQL 表 A 同步到 PostgreSQL 表 B”
Task（子任务）
- 为提升性能，将一个 Job 拆分为多个 Task 并发执行
- 每个 Task 负责同步一部分数据（如若干分表、某一范围分片）
TaskGroup
- 一组 Task 的集合，由框架统一调度执行
- 每个 TaskGroup 内有若干通道（channel），每个 channel 负责一条
```
Reader → Channel → Writer
```
  流水线
Reader 插件
- 数据采集模块，负责从“源数据源”读取数据，发送给框架
- 只关心“如何正确读”，不关注类型转换、指标统计等通用问题
Writer 插件
- 数据写入模块，负责从框架拿数据写入“目标端”
- 只关心“如何正确写”，通用逻辑由框架处理
Transformer（数据转换）
- 可选模块，在 Reader 和 Writer 之间对数据进行转换
- 支持内置 UDF：
```
dx_substr
```
  /
```
dx_pad
```
  /
```
dx_replace
```
  /
```
dx_filter
```
  /
```
dx_groovy
```
- 可以做脱敏、字段裁剪、补全、过滤、自定义 Groovy 脚本转换等
Channel（通道）
- Reader 到 Writer 之间的数据通路和缓冲队列
- 决定并发度 & 流量控制（基于字节数、记录数、通道数）

2.2 架构概览

整体框架：Framework + 插件（Reader / Writer / Transformer）
数据通路（简化）：
- 源端 → Reader → Framework(Channel) → Writer → 目标端
作业生命周期（JobContainer 内部）：
1. ```
preHandler()
```
  – 作业前置处理
2. ```
init()
```
  – 初始化 reader/writer 插件
3. ```
prepare()
```
  – 源端和目标端的准备工作
4. ```
split()
```
  – 按并发度拆分成多个 Task
5. ```
schedule()
```
  – 将 Task 组织为 TaskGroup，并发执行
6. ```
post()
```
  – 全局后置收尾（如 rename 影子表）
7. ```
postHandler()
```
  – 作业后置处理
Task 执行：
- 每个 Task 固定以
```
Reader → Channel → Writer
```
  的线程模型执行
- Channel 内以 Record/Column 为单位传输数据

3. SKILL：与用户交互时的“领域语言”

3.1 如何理解/解释一个 Job JSON

Job JSON 顶层结构：

{
  "job": {
    "settings": {},
    "content": {
      "reader": {},
      "writer": {},
      "transformer": []
    }
  }
}

```
job.settings
```
- 控制本次任务的全局行为
- 重点字段：
  - ```
  speed.byte
```
  ：每秒允许的最大字节数（Bps），
```
  -1
```
  表示不限制
- ```
speed.record
```
    ：每秒允许的最大记录数
  - ```
  speed.channel
```
  ：通道数（影响 Task 数量）
- ```
errorLimit.record
```
    ：允许错误记录总数
  - ```
  errorLimit.percentage
```
  ：允许错误记录占比

job.content.reader

必填，描述数据源及其读取方式

核心字段（以关系型数据库为例）：

{
  "name": "mysqlreader",
  "parameter": {
    "username": "",
    "password": "",
    "column": [],
    "autoPk": false,
    "splitPk": "",
    "connection": [
      {
        "jdbcUrl": [],
        "table": []
      }
    ],
    "where": ""
  }
}

job.content.writer

必填，描述落地目标及写入策略

核心字段（以关系库为例）：

{
  "name": "mysqlwriter",
  "parameter": {
    "username": "",
    "password": "",
    "writeMode": "",
    "column": [],
    "session": [],
    "preSql": [],
    "postSql": [],
    "connection": [
      {
        "jdbcUrl": "",
        "table": []
      }
    ]
  }
}

job.content.transformer

可选，列表形式，每个元素为一个转换规则

典型片段（内置函数）：

{
  "transformer": [
    {
      "name": "dx_substr",
      "parameter": { "idx": 1, "pos": 0, "length": 3 }
    }
  ]
}

AI 在阅读/生成 Job 时，应显式区分：

全局控制（settings） vs 数据流定义（reader/writer） vs 转换规则（transformer）

3.2 任务拆分与并发推导逻辑

用户设定并发度：
```
job.settings.speed.channel = N
```
框架内部：
1. Reader 的
```
split()
```
  按源端特性拆成若干 Task（如按分表、分片、主键范围等）
2. Writer 的
```
split()
```
  需与 Reader 的 Task 数量 1:1 对齐
3. Scheduler 根据
```
taskGroup.channel
```
  （
```
conf/core.json
```
  中配置）决定 TaskGroup 数量：
  - ```
  taskGroupCount = speed.channel / taskGroup.channel
```
与用户讨论“为什么任务这么慢/这么多连接”时，应从：
- ```
speed.channel
```
- Reader 拆分策略（是否按
```
splitPk
```
  / 分区表）
- 目标端写入瓶颈（Writer 能力、批量大小等）入手解释。

3.3 数据质量与错误处理

Addax 在数据质量方面的关键点：

类型不丢失/不失真
- 内部抽象了统一的 Column 类型：
```
Long / Double / String / Date / Timestamp / Bool / Bytes
```
- 每个插件有自己的类型转换策略，保证最小损失
错误控制
- 通过
```
errorLimit.record
```
  和
```
errorLimit.percentage
```
  控制“可容忍错误”
- 超过阈值即认为任务失败
脏数据（Dirty Data）
- 概念：传输过程中因各种原因（例如类型不匹配）导致出错的记录
- 能够过滤、识别、收集与展示脏数据，并统计数量和字节数
- Transformer 层若抛出异常/返回 null，也会影响成功/失败/过滤计数

AI 在帮助用户排错时：

应主动询问/检查：错误是否集中在类型转换、特定列、特定插件
建议合理设置
```
errorLimit
```
，在保障数据质量和任务稳定之间平衡

4. 使用方式与运行环境

4.1 安装与运行

运行时环境
- Java：JDK 17
- Python 2.7+ / 3.7+（仅 Windows 使用本地脚本时需要）
三种典型使用方式

1）Docker 运行示例：

docker pull quay.io/wgzhao/addax:latest
docker run -ti --rm --name addax \
  quay.io/wgzhao/addax:latest \
  /opt/addax/bin/addax.sh /opt/addax/job/job.json

2）一键安装脚本（Linux / macOS）：

/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/wgzhao/Addax/master/install.sh)"

安装目录：
- macOS：
```
/usr/local/addax
```
- Linux：
```
/opt/addax
```

3）源码编译：

git clone https://github.com/wgzhao/addax.git
cd addax
mvn clean package
mvn package -Pdistribution        # 或 assembly:single
# 产出目录示例：target/addax-<version>

首次运行任务

bin/addax.sh job/job.json

4.2 命令行工具 addax.sh

基本用法：

bin/addax.sh <job_file> [options]

关键参数（AI 在给终端命令建议时要正确使用）：

```
-h, --help
```
：帮助
```
-v, --version
```
：版本
```
-l, --log
```
：指定日志文件路径
```
-d, --debug
```
：开启调试模式（IDEA 远程调试会用到）

-L, --log-level

：

DEBUG | INFO | WARN | ERROR

```
-j, --jvm
```
：追加 JVM 参数
```
-p, --params
```
：向 Job 传入动态参数（
```
-Dkey=value
```
形式）

示例（动态参数）：

bin/addax.sh job/test.json \
  -p "-Dusername=root -Dpassword=123456 -Dparam1=value1 -Dparam2=value2"

Job JSON 中可以通过

${param}

访问这些值，比如

${username}

。

内置时间变量（示例时间

2025-07-16 12:13:14

）：

```
${curr_date_short}
```
→
```
20250716
```
```
${curr_date_dash}
```
→
```
2025-07-16
```
```
${curr_datetime_short}
```
→
```
20250716121314
```

${curr_datetime_dash}

→

2025-07-16 12:13:14

${biz_date_short}

${biz_date_dash}

${biz_datetime_*}

等

4.3 Server 模块（HTTP 提交任务）

用途：通过 HTTP 提交 Job JSON 并异步执行，可查询进度和结果
启动脚本：
```
core/src/main/bin/addax-server.sh
```

启动示例：

./addax-server.sh start               # 默认并发上限 30
./addax-server.sh start -p 50 --daemon  # 最大并发 50，后台运行
./addax-server.sh stop

并发配置优先级：

命令行
```
-p
```
/
```
--parallel
```
环境变量
```
ADDAX_SERVER_PARALLEL
```
默认 30

HTTP 接口：

1）提交任务

URL:
```
/api/submit?k1=v1&k2=v2
```
Method: POST
Body: Job JSON

示例：

curl 'http://localhost:10601/api/submit?jobName=example-job' \
  -H 'Content-Type: application/json' \
  -d @job/job.json

响应：

{ "taskId": "xxxx-xxxx-xxxx" }

或并发达到上限时：

{ "error": "ERROR: Maximum number of concurrent tasks reached." }

2）查询任务状态

URL:
```
/api/status?taskId={taskId}
```
Method: GET
响应：

{
  "taskId": "xxxx-xxxx-xxxx",
  "status": "SUCCESS",
  "result": "Job example-job executed.",
  "error": null
}

AI 在设计自动化系统（如调度、工作流）时，可建议用户使用 Server 模块通过 HTTP 集成。

5. 插件系统与二次开发

5.1 支持的数据源（典型）

Reader / Writer 插件覆盖的常见系统包括但不限于：

关系库：MySQL, PostgreSQL, Oracle, SQLServer, SQLite, Greenplum, DB2, Sybase, Doris, StarRocks 等
NoSQL / KV：Cassandra, Redis, MongoDB, HBase（1.x, 2.x，多种模式）
大数据 / 文件：HDFS, Hive, Kudu, Iceberg, Paimon, S3/MinIO, FTP, 本地文件（txt/dbf/excel/json）
时序 / 流：InfluxDB/InfluxDB2, TDengine, Kafka, streamreader/streamwriter
其它：Access, SAP HANA, ClickHouse, Databend 等

当用户问“是否支持 XXX 数据源”时，应：

优先在 docs/reader 与 docs/writer 列表中查找对应
```
xxxreader
```
/
```
xxxwriter
```
若暂不支持，可建议走 插件开发 路线，说明难度和接口模型

5.2 插件开发模型

插件 = 一个 Java 模块 +
```
plugin.json
```
描述 + 若干 jar 依赖
入口类需继承：
- ```
Reader
```
  或
```
Writer
```
  抽象类
- 内部包含两个静态内部类：
```
Job
```
  和
```
Task
```

插件目录结构（约定）：

${ADDAX_HOME}/plugin
  ├── reader
  │   └── <plugin_name>
  │       ├── <plugin_name>-<version>.jar
  │       ├── libs/ -> 指向 shared 依赖目录的符号链接
  │       ├── plugin.json
  │       └── plugin_job_template.json
  └── writer
      └── ...

plugin.json

示例：

{
  "name": "mysqlwriter",
  "class": "com.wgzhao.addax.plugin.writer.mysqlwriter.MysqlWriter",
  "description": "Use Jdbc connect to database, execute insert sql.",
  "developer": "wgzhao"
}

注意：

目录名必须与
```
plugin.json
```
中的
```
name
```
一致
框架通过
```
name
```
找插件，通过
```
class
```
（完全限定名）反射加载

5.3 Job / Task 接口职责（用于代码分析/生成）

以 Reader 为例：

public class SomeReader extends Reader {
    public static class Job extends Reader.Job {
        public void init() { }
        public void prepare() { }
        public List<Configuration> split(int adviceNumber) { return null; }
        public void post() { }
        public void destroy() { }
    }

    public static class Task extends Reader.Task {
        public void init() { }
        public void prepare() { }
        public void startRead(RecordSender recordSender) { }
        public void post() { }
        public void destroy() { }
    }
}

```
Job
```
级别：
- 负责读取插件配置：
```
super.getPluginJobConf()
```
- 全局准备/收尾：如建表/清表、校验权限等
- ```
split(adviceNumber)
```
  ：按并发建议数拆分 Configuration 列表（每个对应一个 Task）

Task

级别：

用
```
super.getPluginJobConf()
```
获取本 Task 的配置
在
```
startRead()
```
/
```
startWrite()
```
中进行实际 I/O
使用
```
RecordSender
```
/
```
RecordReceiver
```
和
```
Record
```
/
```
Column
```
抽象进行传输

重要约束：

Job 和 Task 之间禁止使用共享变量，只能通过配置（Configuration）传递信息
```
prepare
```
/
```
post
```
在 Job 和 Task 层都有，需根据场景选择合适层级实现

5.4 Configuration DSL

Addax 提供

Configuration

类和路径 DSL 来读取 JSON 配置：

路径规则：
- 子对象：
```
a.b.c
```
- 数组元素：
```
a.f[2].g
```

示例 JSON：

{
  "a": {
    "b": { "c": 2 },
    "f": [1, 2, { "g": true }]
  },
  "x": 4
}

示例路径与结果：
- ```
x
```
  →
```
4
```
- ```
a.b.c
```
  →
```
2
```
- ```
a.b.f[2].g
```
  →
```
true
```

AI 在帮用户写插件代码时，可直接给出

Configuration.get("a.b.c")

等示例。

6. 调试、监控与结果上报

6.1 运行日志与调试

本地/远程调试模式：
```
bin/addax.sh -d job/job.json
```
- 程序会以
```
Listening for transport dt_socket at address: 9999
```
  形式暴露 JVM 调试端口
- 可用 IntelliJ IDEA 的 Remote JVM Debug 挂载到指定 host:port

本地调试典型配置：

Main class：
```
com.wgzhao.addax.core.Engine
```

VM Options：

```
-Daddax.home=/opt/app/addax/4.0.3
```
如需加载本地 lib 依赖：
```
-classpath .:/opt/app/addax/4.0.3/lib/*
```

Program arguments:
```
-job job/job.json
```
Working directory:
```
/opt/app/addax/4.0.3
```

6.2 任务运行统计和 Transformer 计量

Addax 会在日志中输出类似：

Total 1000000 records, 22000000 bytes | 
Transform 100000 records(in), 10000 records(out) | 
Speed 2.10MB/s, 100000 records/s | 
Error 0 records, 0 bytes | Percentage 100.00%

以及最终汇总：

任务启动时刻                    : 2015-03-10 17:34:21
任务结束时刻                    : 2015-03-10 17:34:31
任务总计耗时                    :                 10s
任务平均流量                    :            2.10MB/s
记录写入速度                    :         100000rec/s
转换输入总数                    :             1000000
转换输出总数                    :             1000000
读取出记录总数                  :             1000000
同步失败总数                    :                   0

Transformer 维度统计：

输入记录数/字节数
输出记录数/字节数
脏数据记录数/字节数
总耗时

AI 在分析性能问题 / 瓶颈时，应从：

读/写 QPS
Transform 前后记录量变化
错误/脏数据数量
任务耗时与并发度这些维度提示用户。

6.3 任务结果上报（Stats Report）

用途：将 Job 执行的统计结果通过 HTTP POST 报告到外部服务（如监控平台 / 管理端）

配置位置：

$ADDAX_HOME/conf/core.json

→

core.server.address

示例：

{
  "core": {
    "server": {
      "address": "http://localhost:9090/api/v1/addax/jobReport",
      "timeout": 5
    }
  }
}

上报数据结构（JSON）：

{
  "jobName": "test",
  "startTimeStamp": 1587971621,
  "endTimeStamp": 1587971621,
  "totalCosts": 10,
  "totalBytes": 330,
  "byteSpeedPerSecond": 33,
  "recordSpeedPerSecond": 1,
  "totalReadRecords": 6,
  "totalErrorRecords": 0,
  "jobContent": { "配置内容省略": "此处为实际任务配置" }
}

AI 可建议用户：

使用简单 Flask/Node/Java 服务接收并入库统计数据
或与现有监控系统集成

7. 与 Addax 相关的生态项目

addax-admin（后端）
- 仓库：
```
https://github.com/wgzhao/addax-admin
```
- 提供 Web 管理界面和 API，用于管理 Addax 任务
addax-ui（前端）
- 仓库：
```
https://github.com/wgzhao/addax-ui
```
- 为 addax-admin 提供前端展示

AI 在回答“有没有 Web 管理界面 / 调度平台”时，可推荐这两个项目。

8. 版本与贡献

版本规范：遵循 SemVer (
```
x.y.z
```
)
- ```
z
```
  Patch：兼容修复、性能优化
- ```
y
```
  Minor：新增特性或兼容性风险较小的修改
- ```
x
```
  Major：重大变更，通常不向后兼容
开发规范（概略）
- 使用 IntelliJ + Airlift 代码风格
- 异常使用
```
AddaxException
```
  ，并区分错误类型
- 谨慎使用 Stream API，避免在性能敏感路径中滥用
- 避免复杂三元表达式
- 所有文件需包含 Apache 2.0 许可证头
- Commit message 参考 https://cbea.ms/git-commit/

AI 在生成 PR/代码建议时，应尽量贴合以上风格。

SKILL: Addax 项目知识

1. 项目整体认识

1.1 项目定位

名称：Addax
类型：通用开源 ETL 工具（Extract–Transform–Load）
起源：基于阿里巴巴 DataX 的 fork 与演进
目标：在多种异构数据源之间，提供稳定、高效、可扩展的“离线数据同步”能力

1.2 核心价值

支持 20+ SQL/NoSQL/文件/时序/大数据 数据源
使用 JSON 任务配置 即可完成复杂同步，无需写代码
插件化架构，Reader / Writer / Transformer 解耦，可自由扩展
提供 数据质量监控、速率控制、错误容忍、脏数据探测 等生产级能力
既可命令行运行，也可通过 Server 模块 HTTP 接口 异步提交和管理任务
有配套的 addax-admin / addax-ui 项目做 Web 管控

2. 概念与架构模型

2.1 核心业务概念

在与用户讨论 / 理解需求时，应优先按以下抽象模型理解：

Job（作业）
- 一次完整的数据同步任务，从一个源到一个目标
- 通过一个 JSON 文件描述：数据源 reader、目标端 writer、变换规则、速率控制、错误阈值等
- Job 是业务上的最小单位，如 “从 MySQL 表 A 同步到 PostgreSQL 表 B”
Task（子任务）
- 为提升性能，将一个 Job 拆分为多个 Task 并发执行
- 每个 Task 负责同步一部分数据（如若干分表、某一范围分片）
TaskGroup
- 一组 Task 的集合，由框架统一调度执行
- 每个 TaskGroup 内有若干通道（channel），每个 channel 负责一条
```
Reader → Channel → Writer
```
  流水线
Reader 插件
- 数据采集模块，负责从“源数据源”读取数据，发送给框架
- 只关心“如何正确读”，不关注类型转换、指标统计等通用问题
Writer 插件
- 数据写入模块，负责从框架拿数据写入“目标端”
- 只关心“如何正确写”，通用逻辑由框架处理
Transformer（数据转换）
- 可选模块，在 Reader 和 Writer 之间对数据进行转换
- 支持内置 UDF：
```
dx_substr
```
  /
```
dx_pad
```
  /
```
dx_replace
```
  /
```
dx_filter
```
  /
```
dx_groovy
```
- 可以做脱敏、字段裁剪、补全、过滤、自定义 Groovy 脚本转换等
Channel（通道）
- Reader 到 Writer 之间的数据通路和缓冲队列
- 决定并发度 & 流量控制（基于字节数、记录数、通道数）

2.2 架构概览

整体框架：Framework + 插件（Reader / Writer / Transformer）
数据通路（简化）：
- 源端 → Reader → Framework(Channel) → Writer → 目标端
作业生命周期（JobContainer 内部）：
1. ```
preHandler()
```
  – 作业前置处理
2. ```
init()
```
  – 初始化 reader/writer 插件
3. ```
prepare()
```
  – 源端和目标端的准备工作
4. ```
split()
```
  – 按并发度拆分成多个 Task
5. ```
schedule()
```
  – 将 Task 组织为 TaskGroup，并发执行
6. ```
post()
```
  – 全局后置收尾（如 rename 影子表）
7. ```
postHandler()
```
  – 作业后置处理
Task 执行：
- 每个 Task 固定以
```
Reader → Channel → Writer
```
  的线程模型执行
- Channel 内以 Record/Column 为单位传输数据

3. SKILL：与用户交互时的“领域语言”

3.1 如何理解/解释一个 Job JSON

Job JSON 顶层结构：

{
  "job": {
    "settings": {},
    "content": {
      "reader": {},
      "writer": {},
      "transformer": []
    }
  }
}

```
job.settings
```
- 控制本次任务的全局行为
- 重点字段：
  - ```
  speed.byte
```
  ：每秒允许的最大字节数（Bps），
```
  -1
```
  表示不限制
- ```
speed.record
```
    ：每秒允许的最大记录数
  - ```
  speed.channel
```
  ：通道数（影响 Task 数量）
- ```
errorLimit.record
```
    ：允许错误记录总数
  - ```
  errorLimit.percentage
```
  ：允许错误记录占比

job.content.reader

必填，描述数据源及其读取方式

核心字段（以关系型数据库为例）：

{
  "name": "mysqlreader",
  "parameter": {
    "username": "",
    "password": "",
    "column": [],
    "autoPk": false,
    "splitPk": "",
    "connection": [
      {
        "jdbcUrl": [],
        "table": []
      }
    ],
    "where": ""
  }
}

job.content.writer

必填，描述落地目标及写入策略

核心字段（以关系库为例）：

{
  "name": "mysqlwriter",
  "parameter": {
    "username": "",
    "password": "",
    "writeMode": "",
    "column": [],
    "session": [],
    "preSql": [],
    "postSql": [],
    "connection": [
      {
        "jdbcUrl": "",
        "table": []
      }
    ]
  }
}

job.content.transformer

可选，列表形式，每个元素为一个转换规则

典型片段（内置函数）：

{
  "transformer": [
    {
      "name": "dx_substr",
      "parameter": { "idx": 1, "pos": 0, "length": 3 }
    }
  ]
}

AI 在阅读/生成 Job 时，应显式区分：

全局控制（settings） vs 数据流定义（reader/writer） vs 转换规则（transformer）

3.2 任务拆分与并发推导逻辑

用户设定并发度：
```
job.settings.speed.channel = N
```
框架内部：
1. Reader 的
```
split()
```
  按源端特性拆成若干 Task（如按分表、分片、主键范围等）
2. Writer 的
```
split()
```
  需与 Reader 的 Task 数量 1:1 对齐
3. Scheduler 根据
```
taskGroup.channel
```
  （
```
conf/core.json
```
  中配置）决定 TaskGroup 数量：
  - ```
  taskGroupCount = speed.channel / taskGroup.channel
```
与用户讨论“为什么任务这么慢/这么多连接”时，应从：
- ```
speed.channel
```
- Reader 拆分策略（是否按
```
splitPk
```
  / 分区表）
- 目标端写入瓶颈（Writer 能力、批量大小等）入手解释。

3.3 数据质量与错误处理

Addax 在数据质量方面的关键点：

类型不丢失/不失真
- 内部抽象了统一的 Column 类型：
```
Long / Double / String / Date / Timestamp / Bool / Bytes
```
- 每个插件有自己的类型转换策略，保证最小损失
错误控制
- 通过
```
errorLimit.record
```
  和
```
errorLimit.percentage
```
  控制“可容忍错误”
- 超过阈值即认为任务失败
脏数据（Dirty Data）
- 概念：传输过程中因各种原因（例如类型不匹配）导致出错的记录
- 能够过滤、识别、收集与展示脏数据，并统计数量和字节数
- Transformer 层若抛出异常/返回 null，也会影响成功/失败/过滤计数

AI 在帮助用户排错时：

应主动询问/检查：错误是否集中在类型转换、特定列、特定插件
建议合理设置
```
errorLimit
```
，在保障数据质量和任务稳定之间平衡

4. 使用方式与运行环境

4.1 安装与运行

运行时环境
- Java：JDK 17
- Python 2.7+ / 3.7+（仅 Windows 使用本地脚本时需要）
三种典型使用方式

1）Docker 运行示例：

docker pull quay.io/wgzhao/addax:latest
docker run -ti --rm --name addax \
  quay.io/wgzhao/addax:latest \
  /opt/addax/bin/addax.sh /opt/addax/job/job.json

2）一键安装脚本（Linux / macOS）：

/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/wgzhao/Addax/master/install.sh)"

安装目录：
- macOS：
```
/usr/local/addax
```
- Linux：
```
/opt/addax
```

3）源码编译：

git clone https://github.com/wgzhao/addax.git
cd addax
mvn clean package
mvn package -Pdistribution        # 或 assembly:single
# 产出目录示例：target/addax-<version>

首次运行任务

bin/addax.sh job/job.json

4.2 命令行工具 addax.sh

基本用法：

bin/addax.sh <job_file> [options]

关键参数（AI 在给终端命令建议时要正确使用）：

```
-h, --help
```
：帮助
```
-v, --version
```
：版本
```
-l, --log
```
：指定日志文件路径
```
-d, --debug
```
：开启调试模式（IDEA 远程调试会用到）

-L, --log-level

：

DEBUG | INFO | WARN | ERROR

```
-j, --jvm
```
：追加 JVM 参数
```
-p, --params
```
：向 Job 传入动态参数（
```
-Dkey=value
```
形式）

示例（动态参数）：

bin/addax.sh job/test.json \
  -p "-Dusername=root -Dpassword=123456 -Dparam1=value1 -Dparam2=value2"

Job JSON 中可以通过

${param}

访问这些值，比如

${username}

。

内置时间变量（示例时间

2025-07-16 12:13:14

）：

```
${curr_date_short}
```
→
```
20250716
```
```
${curr_date_dash}
```
→
```
2025-07-16
```
```
${curr_datetime_short}
```
→
```
20250716121314
```

${curr_datetime_dash}

→

2025-07-16 12:13:14

${biz_date_short}

${biz_date_dash}

${biz_datetime_*}

等

4.3 Server 模块（HTTP 提交任务）

用途：通过 HTTP 提交 Job JSON 并异步执行，可查询进度和结果
启动脚本：
```
core/src/main/bin/addax-server.sh
```

启动示例：

./addax-server.sh start               # 默认并发上限 30
./addax-server.sh start -p 50 --daemon  # 最大并发 50，后台运行
./addax-server.sh stop

并发配置优先级：

命令行
```
-p
```
/
```
--parallel
```
环境变量
```
ADDAX_SERVER_PARALLEL
```
默认 30

HTTP 接口：

1）提交任务

URL:
```
/api/submit?k1=v1&k2=v2
```
Method: POST
Body: Job JSON

示例：

curl 'http://localhost:10601/api/submit?jobName=example-job' \
  -H 'Content-Type: application/json' \
  -d @job/job.json

响应：

{ "taskId": "xxxx-xxxx-xxxx" }

或并发达到上限时：

{ "error": "ERROR: Maximum number of concurrent tasks reached." }

2）查询任务状态

URL:
```
/api/status?taskId={taskId}
```
Method: GET
响应：

{
  "taskId": "xxxx-xxxx-xxxx",
  "status": "SUCCESS",
  "result": "Job example-job executed.",
  "error": null
}

AI 在设计自动化系统（如调度、工作流）时，可建议用户使用 Server 模块通过 HTTP 集成。

5. 插件系统与二次开发

5.1 支持的数据源（典型）

Reader / Writer 插件覆盖的常见系统包括但不限于：

关系库：MySQL, PostgreSQL, Oracle, SQLServer, SQLite, Greenplum, DB2, Sybase, Doris, StarRocks 等
NoSQL / KV：Cassandra, Redis, MongoDB, HBase（1.x, 2.x，多种模式）
大数据 / 文件：HDFS, Hive, Kudu, Iceberg, Paimon, S3/MinIO, FTP, 本地文件（txt/dbf/excel/json）
时序 / 流：InfluxDB/InfluxDB2, TDengine, Kafka, streamreader/streamwriter
其它：Access, SAP HANA, ClickHouse, Databend 等

当用户问“是否支持 XXX 数据源”时，应：

优先在 docs/reader 与 docs/writer 列表中查找对应
```
xxxreader
```
/
```
xxxwriter
```
若暂不支持，可建议走 插件开发 路线，说明难度和接口模型

5.2 插件开发模型

插件 = 一个 Java 模块 +
```
plugin.json
```
描述 + 若干 jar 依赖
入口类需继承：
- ```
Reader
```
  或
```
Writer
```
  抽象类
- 内部包含两个静态内部类：
```
Job
```
  和
```
Task
```

插件目录结构（约定）：

${ADDAX_HOME}/plugin
  ├── reader
  │   └── <plugin_name>
  │       ├── <plugin_name>-<version>.jar
  │       ├── libs/ -> 指向 shared 依赖目录的符号链接
  │       ├── plugin.json
  │       └── plugin_job_template.json
  └── writer
      └── ...

plugin.json

示例：

{
  "name": "mysqlwriter",
  "class": "com.wgzhao.addax.plugin.writer.mysqlwriter.MysqlWriter",
  "description": "Use Jdbc connect to database, execute insert sql.",
  "developer": "wgzhao"
}

注意：

目录名必须与
```
plugin.json
```
中的
```
name
```
一致
框架通过
```
name
```
找插件，通过
```
class
```
（完全限定名）反射加载

5.3 Job / Task 接口职责（用于代码分析/生成）

以 Reader 为例：

public class SomeReader extends Reader {
    public static class Job extends Reader.Job {
        public void init() { }
        public void prepare() { }
        public List<Configuration> split(int adviceNumber) { return null; }
        public void post() { }
        public void destroy() { }
    }

    public static class Task extends Reader.Task {
        public void init() { }
        public void prepare() { }
        public void startRead(RecordSender recordSender) { }
        public void post() { }
        public void destroy() { }
    }
}

```
Job
```
级别：
- 负责读取插件配置：
```
super.getPluginJobConf()
```
- 全局准备/收尾：如建表/清表、校验权限等
- ```
split(adviceNumber)
```
  ：按并发建议数拆分 Configuration 列表（每个对应一个 Task）

Task

级别：

用
```
super.getPluginJobConf()
```
获取本 Task 的配置
在
```
startRead()
```
/
```
startWrite()
```
中进行实际 I/O
使用
```
RecordSender
```
/
```
RecordReceiver
```
和
```
Record
```
/
```
Column
```
抽象进行传输

重要约束：

Job 和 Task 之间禁止使用共享变量，只能通过配置（Configuration）传递信息
```
prepare
```
/
```
post
```
在 Job 和 Task 层都有，需根据场景选择合适层级实现

5.4 Configuration DSL

Addax 提供

Configuration

类和路径 DSL 来读取 JSON 配置：

路径规则：
- 子对象：
```
a.b.c
```
- 数组元素：
```
a.f[2].g
```

示例 JSON：

{
  "a": {
    "b": { "c": 2 },
    "f": [1, 2, { "g": true }]
  },
  "x": 4
}

示例路径与结果：
- ```
x
```
  →
```
4
```
- ```
a.b.c
```
  →
```
2
```
- ```
a.b.f[2].g
```
  →
```
true
```

AI 在帮用户写插件代码时，可直接给出

Configuration.get("a.b.c")

等示例。

6. 调试、监控与结果上报

6.1 运行日志与调试

本地/远程调试模式：
```
bin/addax.sh -d job/job.json
```
- 程序会以
```
Listening for transport dt_socket at address: 9999
```
  形式暴露 JVM 调试端口
- 可用 IntelliJ IDEA 的 Remote JVM Debug 挂载到指定 host:port

本地调试典型配置：

Main class：
```
com.wgzhao.addax.core.Engine
```

VM Options：

```
-Daddax.home=/opt/app/addax/4.0.3
```
如需加载本地 lib 依赖：
```
-classpath .:/opt/app/addax/4.0.3/lib/*
```

Program arguments:
```
-job job/job.json
```
Working directory:
```
/opt/app/addax/4.0.3
```

6.2 任务运行统计和 Transformer 计量

Addax 会在日志中输出类似：

Total 1000000 records, 22000000 bytes | 
Transform 100000 records(in), 10000 records(out) | 
Speed 2.10MB/s, 100000 records/s | 
Error 0 records, 0 bytes | Percentage 100.00%

以及最终汇总：

任务启动时刻                    : 2015-03-10 17:34:21
任务结束时刻                    : 2015-03-10 17:34:31
任务总计耗时                    :                 10s
任务平均流量                    :            2.10MB/s
记录写入速度                    :         100000rec/s
转换输入总数                    :             1000000
转换输出总数                    :             1000000
读取出记录总数                  :             1000000
同步失败总数                    :                   0

Transformer 维度统计：

输入记录数/字节数
输出记录数/字节数
脏数据记录数/字节数
总耗时

AI 在分析性能问题 / 瓶颈时，应从：

读/写 QPS
Transform 前后记录量变化
错误/脏数据数量
任务耗时与并发度这些维度提示用户。

6.3 任务结果上报（Stats Report）

用途：将 Job 执行的统计结果通过 HTTP POST 报告到外部服务（如监控平台 / 管理端）

配置位置：

$ADDAX_HOME/conf/core.json

→

core.server.address

示例：

{
  "core": {
    "server": {
      "address": "http://localhost:9090/api/v1/addax/jobReport",
      "timeout": 5
    }
  }
}

上报数据结构（JSON）：

{
  "jobName": "test",
  "startTimeStamp": 1587971621,
  "endTimeStamp": 1587971621,
  "totalCosts": 10,
  "totalBytes": 330,
  "byteSpeedPerSecond": 33,
  "recordSpeedPerSecond": 1,
  "totalReadRecords": 6,
  "totalErrorRecords": 0,
  "jobContent": { "配置内容省略": "此处为实际任务配置" }
}

AI 可建议用户：

使用简单 Flask/Node/Java 服务接收并入库统计数据
或与现有监控系统集成

7. 与 Addax 相关的生态项目

addax-admin（后端）
- 仓库：
```
https://github.com/wgzhao/addax-admin
```
- 提供 Web 管理界面和 API，用于管理 Addax 任务
addax-ui（前端）
- 仓库：
```
https://github.com/wgzhao/addax-ui
```
- 为 addax-admin 提供前端展示

AI 在回答“有没有 Web 管理界面 / 调度平台”时，可推荐这两个项目。

8. 版本与贡献

版本规范：遵循 SemVer (
```
x.y.z
```
)
- ```
z
```
  Patch：兼容修复、性能优化
- ```
y
```
  Minor：新增特性或兼容性风险较小的修改
- ```
x
```
  Major：重大变更，通常不向后兼容
开发规范（概略）
- 使用 IntelliJ + Airlift 代码风格
- 异常使用
```
AddaxException
```
  ，并区分错误类型
- 谨慎使用 Stream API，避免在性能敏感路径中滥用
- 避免复杂三元表达式
- 所有文件需包含 Apache 2.0 许可证头
- Commit message 参考 https://cbea.ms/git-commit/

AI 在生成 PR/代码建议时，应尽量贴合以上风格。

SKILL: Addax 项目知识

Related Skills

Markdown Converter

Nano Banana Pro

1password