本章节详细介绍使用 addax 进行数据迁移的操作过程。
准备工作
操作系统:TencentOS Server。
规格:内存至少2GB。
已安装 JDK 1.8 及以上版本。
步骤1: 上传迁移工具
将校验工具 addax 的工具包上传至 CVM 上。
addax 的工具包目录结构如下。
步骤2:迁移任务配置
1. 执行 vim 命令,打开 job 文件夹下的迁移任务配置文件 ctsdb4es2influxdb.json。
vim job/ctsdb4es2influxdb.json
2. 在 ctsdb4es2influxdb.json 文件中,参照如下表所列的参数介绍,配置迁移迁移读写信息。具体参数,请参见下表所示。
{"job": {"setting": {"speed": {"byte": -1,"channel": 1}},"content": {"reader": {"name": "ctsdb4esreader","parameter": {"endpoint": "http://127.0.0.1:9200","accessId": "","accesskey": "","index": "test","type": "","searchType": "query_then_fetch","headers": {},"scroll": "10m","timeType": 3,"search": [{"size": 3000,"query": {"range": {"timestamp": {"gte": 1692982800000,"lt": 1692983400000}}},"docvalue_fields":["timestamp","agent.hostname","agent.id","cpuUsage","region"]}],"column": ["timestamp","agent.hostname","agent.id","cpuUsage","region"]}},"writer": {"name": "influxdb2writer","parameter": {"connection": [{"endpoint": "http://127.0.0.1:10002","bucket": "ctsdb4es","org": "com.wgzhao","table": "test"}],"username": "","passwd": "","tag": ["region"],"interval": "ns","column": ["agent.hostname","agent.id","cpuUsage","region"],"batchSize": 1000}}}}}
说明:
CTSDB3.0支持的数据类型为 Float,Integer,String,Boolean,Timestamp,其余类型并不支持。如果要迁移的数据列是对象,则需要进行展开,例如 agent.hostname, agent.id 的方式,将对象展开成多个列进行迁移。
读写方式 | 参数名 | 参数说明 |
read (CTSDB 1.0与2.0) | name | 固定值 ctsdb4esreader。 |
| endpoint | ctsdb 1.0 或 2.0 for es 的访问地址。 |
| accessId | 认证使用的 accessId。 |
| accesskey | 认证使用的 accessKey。 |
| index | 要迁移的 index。 |
| type | 要迁移的 Index type。 |
| searchType | 固定值 query_then_fetch。 |
| headers | 请求 CTSDB2.0携带的请求头。 |
| scroll | 滚动查询的缓存时间。 |
| timeType | 时间列类型,有1,2,3,4四种。 第一种的格式(ISO-8601时间)是2024-09-03T11:53:44.120。 第二种的格式(秒级时间戳)是1725865926。 第三种的格式(毫秒时间戳)是1725865926000。 第四种的时间格式(年月日)是20240909。 |
| search | 要迁移数据的查询语句,如果要迁移的数据列是对象,则需要进行展开。例如 agent.hostname, agent.id。 |
| column | 要迁移的列,第一列必须是时间列,列名需要和查询出的列名一致。 |
write(CTSDB 3.0) | name | 固定值 influxdb2writer。 |
| endpoint | |
| org | 固定值 com.wgzhao。 |
| bucket | CTSDB 3.0 数据库 Database 名。迁移前,请准备好数据库。 |
| table | 迁移到的表名。 |
| username | CTSDB 3.0 的实例名,ctsdbi- 开头。 |
| passwd | CTSDB 3.0 实例的访问密码。 |
| tag | 配置哪些列为 tag。 |
| interval | 固定值,时间单位 ns 纳秒。 |
| column | 迁移的列,要和 reader 配置的顺序相同,且不包含时间列。 |
| batchSize | 每批写入 influxdb 批量大小,建议是 reader 的10倍。 |
| bytes | -1默认不限制,"bytes" 单位,如速度限制为 1048576 字节(即 1MB)。 |
| channel | 固定值1。 |
步骤3:启动迁移
1. 进入 Addax 目录下,执行如下命令,启动迁移任务。
cd addax-4.1.5-SNAPSHOTbin/addax.sh job/ctsdb4es2influxdb.json
显示如下信息,说明数据正在迁移中。
2. 等待迁移任务执行完成。
显示如下信息,Failed record 为失败的条数,该值为0,说明全部迁移成功。
说明:
如果有异常,比如读取失败,转换失败,写入冲突,写入失败等,会直接中断迁移,非中断的异常可忽略,如:
ClassNotFoundException:org.fusesource.jansi.AnsiConsole
。如果读取或写入操作在任何时候遇到失败,那么从出错的那一行开始,之后所有的数据导入都将终止;相反,如果整个操作过程中没有出现任何失败,那么所有数据都将被成功导入。