datax-xugudb-adapter

name	datax-xugudb-adapter
description	阿里巴巴 DataX 数据同步工具适配虚谷数据库（XuguDB）的完整指南。当用户需要将基于 DataX 的数据同步工具配置或适配到虚谷数据库时使用此技能，包括安装部署、插件配置、任务配置、性能优化等。适用于数据同步、数据迁移、数据集成等场景。

阿里巴巴 DataX 数据同步工具虚谷数据库适配指南

概述

本技能提供阿里巴巴 DataX 数据同步工具适配虚谷数据库（XuguDB）的完整配置指南。DataX 是阿里巴巴开源的数据同步工具，支持多种数据源之间的数据同步，现在可以通过配置支持虚谷数据库作为数据源或目标数据库。

适用场景：

• 数据同步
• 数据迁移
• 数据集成
• 数据仓库建设
• 数据备份

核心特性：

• 支持多种数据源
• 高性能数据同步
• 支持全量/增量同步
• 支持数据转换
• 支持断点续传
• 支持集群部署

快速开始

1. 安装部署

下载 DataX：

# 下载 DataX
wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/202309/datax.tar.gz

# 解压
tar -xzf datax.tar.gz
cd datax

# 验证安装
python bin/datax.py --version

使用 Docker 部署：

# 拉取镜像
docker pull datax/datax:latest

# 运行容器
docker run -d --name datax \
  -v /path/to/jobs:/jobs \
  datax/datax:latest

2. 配置虚谷数据库插件

添加虚谷数据库插件：

# 创建虚谷数据库 reader 插件目录
mkdir -p plugin/reader/xugudbreader

# 创建虚谷数据库 writer 插件目录
mkdir -p plugin/writer/xugudbwriter

# 复制插件文件
cp xugudbreader/* plugin/reader/xugudbreader/
cp xugudbwriter/* plugin/writer/xugudbwriter/

# 复制 JDBC 驱动
cp xugu-jdbc-*.jar plugin/reader/xugudbreader/libs/
cp xugu-jdbc-*.jar plugin/writer/xugudbwriter/libs/

3. 创建第一个同步任务

创建任务配置文件 xugudb2xugudb.json：

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "xugudbreader",
                    "parameter": {
                        "column": ["id", "name", "email", "phone", "status"],
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8"],
                                "table": ["users"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA",
                        "where": "status = 1"
                    }
                },
                "writer": {
                    "name": "xugudbwriter",
                    "parameter": {
                        "column": ["id", "name", "email", "phone", "status"],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8",
                                "table": ["users_target"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA",
                        "writeMode": "insert"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 3
            }
        }
    }
}

执行同步任务：

# 执行任务
python bin/datax.py job/xugudb2xugudb.json

# 查看任务日志
tail -f log/datax.log

数据源配置

1. Reader 配置

虚谷数据库 Reader 配置：

{
    "name": "xugudbreader",
    "parameter": {
        "username": "SYSDBA",
        "password": "SYSDBA",
        "column": [
            "id",
            "name",
            "email",
            "phone",
            "status",
            "created_at"
        ],
        "connection": [
            {
                "table": [
                    "users"
                ],
                "jdbcUrl": [
                    "jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8"
                ]
            }
        ],
        "where": "status = 1 AND created_at >= '2024-01-01'",
        "querySql": [
            "SELECT id, name, email, phone, status, created_at FROM users WHERE status = 1"
        ],
        "fetchSize": 1000,
        "splitPk": "id",
        "splitStrategy": "MOD",
        "haveKerberos": false,
        "kerberosKeytabFilePath": "",
        "kerberosPrincipal": ""
    }
}

2. Writer 配置

虚谷数据库 Writer 配置：

{
    "name": "xugudbwriter",
    "parameter": {
        "username": "SYSDBA",
        "password": "SYSDBA",
        "column": [
            "id",
            "name",
            "email",
            "phone",
            "status",
            "created_at"
        ],
        "connection": [
            {
                "table": [
                    "users_target"
                ],
                "jdbcUrl": "jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8"
            }
        ],
        "writeMode": "insert",
        "batchSize": 1000,
        "preSql": [
            "DELETE FROM users_target WHERE created_at < DATE_SUB(NOW(), INTERVAL 30 DAY)"
        ],
        "postSql": [
            "UPDATE users_target SET sync_time = NOW() WHERE sync_time IS NULL"
        ],
        "session": [
            "SET SESSION group_concat_max_len = 102400"
        ]
    }
}

3. 高级配置

连接池配置：

{
    "reader": {
        "name": "xugudbreader",
        "parameter": {
            "connection": [
                {
                    "jdbcUrl": [
                        "jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8"
                    ],
                    "table": ["users"]
                }
            ],
            "username": "SYSDBA",
            "password": "SYSDBA",
            "column": ["*"],
            "fetchSize": 1000,
            "where": "",
            "querySql": [],
            "splitPk": "id",
            "splitStrategy": "MOD",
            "haveKerberos": false,
            "kerberosKeytabFilePath": "",
            "kerberosPrincipal": ""
        }
    }
}

任务配置

1. 全量同步

全量同步配置：

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "xugudbreader",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8"],
                                "table": ["users"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA"
                    }
                },
                "writer": {
                    "name": "xugudbwriter",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8",
                                "table": ["users_target"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA",
                        "writeMode": "insert",
                        "preSql": ["DELETE FROM users_target"]
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 3
            }
        }
    }
}

2. 增量同步

增量同步配置：

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "xugudbreader",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8"],
                                "table": ["users"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA",
                        "where": "updated_at > '${last_sync_time}'"
                    }
                },
                "writer": {
                    "name": "xugudbwriter",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8",
                                "table": ["users_target"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA",
                        "writeMode": "replace"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 3
            }
        }
    }
}

3. 分库分表同步

分库分表同步配置：

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "xugudbreader",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": [
                                    "jdbc:xugu://192.168.1.101:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8",
                                    "jdbc:xugu://192.168.1.102:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8",
                                    "jdbc:xugu://192.168.1.103:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8"
                                ],
                                "table": ["users_0", "users_1", "users_2"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA"
                    }
                },
                "writer": {
                    "name": "xugudbwriter",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:xugu://192.168.1.200:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8",
                                "table": ["users"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA",
                        "writeMode": "insert"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 9
            }
        }
    }
}

性能优化

1. 通道配置

通道数量优化：

{
    "job": {
        "setting": {
            "speed": {
                "channel": 5,
                "byte": 10485760,
                "record": 10000
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.02
            }
        }
    }
}

2. 批量处理配置

批量大小优化：

{
    "reader": {
        "name": "xugudbreader",
        "parameter": {
            "fetchSize": 1000,
            "queryTimeout": 30000
        }
    },
    "writer": {
        "name": "xugudbwriter",
        "parameter": {
            "batchSize": 1000,
            "batchTimeout": 1000,
            "writeMode": "insert"
        }
    }
}

3. 内存优化

JVM 内存配置：

# 修改 DataX 启动脚本
# 编辑 bin/datax.py

# 设置 JVM 内存参数
JVM_XMS = "2g"
JVM_XMX = "4g"
JVM_XMN = "1g"
JVM_MaxPermSize = "256m"

# 或者通过环境变量设置
export DATAX_JAVA_OPTS="-Xms2g -Xmx4g -Xmn1g -XX:MaxPermSize=256m"

4. 数据库优化

数据库连接优化：

{
    "reader": {
        "name": "xugudbreader",
        "parameter": {
            "connection": [
                {
                    "jdbcUrl": [
                        "jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8&connectionTimeout=30000&socketTimeout=60000&autoReconnect=true"
                    ],
                    "table": ["users"]
                }
            ],
            "fetchSize": 1000,
            "queryTimeout": 30000
        }
    }
}

数据转换

1. 字段映射

字段映射配置：

{
    "reader": {
        "name": "xugudbreader",
        "parameter": {
            "column": [
                "id",
                "name",
                "email",
                "phone",
                "status",
                "created_at"
            ]
        }
    },
    "writer": {
        "name": "xugudbwriter",
        "parameter": {
            "column": [
                "user_id",
                "user_name",
                "user_email",
                "user_phone",
                "user_status",
                "create_time"
            ]
        }
    }
}

2. 数据类型转换

数据类型映射：

{
    "transformer": [
        {
            "name": "dx_substr",
            "parameter": {
                "columnIndex": 1,
                "paras": ["0", "10"]
            }
        },
        {
            "name": "dx_replace",
            "parameter": {
                "columnIndex": 2,
                "paras": ["old", "new"]
            }
        },
        {
            "name": "dx_filter",
            "parameter": {
                "columnIndex": 4,
                "paras": ["greaterThan", "0"]
            }
        }
    ]
}

3. 数据过滤

数据过滤配置：

{
    "reader": {
        "name": "xugudbreader",
        "parameter": {
            "where": "status = 1 AND created_at >= '2024-01-01'"
        }
    },
    "transformer": [
        {
            "name": "dx_groovy",
            "parameter": {
                "code": "record.setColumnValue(1, record.getColumnValue(1).toUpperCase())",
                "extraPackage": []
            }
        }
    ]
}

集群部署

1. 集群配置

集群节点配置：

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "xugudbreader",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8"],
                                "table": ["users"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA",
                        "splitPk": "id",
                        "splitStrategy": "MOD"
                    }
                },
                "writer": {
                    "name": "xugudbwriter",
                    "parameter": {
                        "column": ["*"],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:xugu://127.0.0.1:5138/SYSTEM?current_schema=SYSDBA&CHAR_SET=UTF8",
                                "table": ["users_target"]
                            }
                        ],
                        "password": "SYSDBA",
                        "username": "SYSDBA",
                        "writeMode": "insert"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 10
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.02
            }
        }
    }
}

2. 负载均衡配置

负载均衡策略：

{
    "job": {
        "setting": {
            "speed": {
                "channel": 10,
                "byte": 104857600,
                "record": 100000
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.02
            },
            "loadBalance": {
                "strategy": "round_robin",
                "maxParallelTasks": 5
            }
        }
    }
}

监控管理

1. 日志配置

日志级别配置：

<!-- log4j.xml -->
<configuration>
    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>
    
    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>log/datax.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>log/datax.%d{yyyy-MM-dd}.log</fileNamePattern>
            <maxHistory>30</maxHistory>
        </rollingPolicy>
        <encoder>
            <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>
    
    <logger name="com.alibaba.datax" level="INFO"/>
    <logger name="org.apache" level="WARN"/>
    
    <root level="INFO">
        <appender-ref ref="STDOUT"/>
        <appender-ref ref="FILE"/>
    </root>
</configuration>

2. 性能监控

监控脚本：

#!/bin/bash

# 监控 DataX 任务
monitor_datax_job() {
    local job_id=$1
    local log_file="log/datax_${job_id}.log"
    
    # 检查任务状态
    if grep -q "任务完成" "$log_file"; then
        echo "任务 $job_id 完成"
        return 0
    elif grep -q "任务失败" "$log_file"; then
        echo "任务 $job_id 失败"
        return 1
    else
        echo "任务 $job_id 执行中"
        return 2
    fi
}

# 监控所有任务
monitor_all_jobs() {
    for job_file in job/*.json; do
        local job_id=$(basename "$job_file" .json)
        monitor_datax_job "$job_id"
    done
}

# 发送报警
send_alert() {
    local message=$1
    echo "发送报警: $message"
    # 发送邮件或其他通知
}

# 主函数
main() {
    while true; do
        monitor_all_jobs
        sleep 60
    done
}

main

3. 任务调度

使用 cron 调度：

# 编辑 crontab
crontab -e

# 添加定时任务
# 每天凌晨 2 点执行数据同步
0 2 * * * /path/to/datax/bin/datax.py /path/to/jobs/daily_sync.json >> /path/to/logs/daily_sync.log 2>&1

# 每小时执行增量同步
0 * * * * /path/to/datax/bin/datax.py /path/to/jobs/incremental_sync.json >> /path/to/logs/incremental_sync.log 2>&1

故障排除

1. 连接问题

问题：无法连接到虚谷数据库。

解决方案：

# 检查数据库服务状态
systemctl status xugudb

# 检查连接配置
cat job/*.json | grep -i xugu

# 测试数据库连接
java -cp "lib/*:plugin/reader/xugudbreader/libs/*" com.xugu.cloudjdbc.Driver -url "jdbc:xugu://127.0.0.1:5138/SYSTEM" -user SYSDBA -password SYSDBA

# 检查网络连接
ping 127.0.0.1
telnet 127.0.0.1 5138

2. 性能问题

问题：数据同步缓慢。

解决方案：

# 检查系统资源
top
free -h
df -h

# 检查数据库性能
# 在虚谷数据库中执行
SELECT * FROM v$session WHERE status = 'ACTIVE';
SELECT * FROM v$sql ORDER BY elapsed_time DESC;

# 优化 JVM 参数
export DATAX_JAVA_OPTS="-Xms4g -Xmx8g -Xmn2g -XX:MaxPermSize=512m"

# 优化通道数量
# 在任务配置中增加通道数量

3. 内存问题

问题：内存溢出。

解决方案：

# 增加 JVM 内存
export DATAX_JAVA_OPTS="-Xms8g -Xmx16g -Xmn4g -XX:MaxPermSize=1g"

# 优化内存使用
# 1. 减少批量大小
# 2. 增加通道数量
# 3. 使用流式处理

# 监控内存使用
jstat -gcutil <pid> 1000

4. 数据类型问题

问题：数据类型转换错误。

解决方案：

{
    "reader": {
        "name": "xugudbreader",
        "parameter": {
            "column": [
                {
                    "name": "id",
                    "type": "long"
                },
                {
                    "name": "name",
                    "type": "string"
                },
                {
                    "name": "amount",
                    "type": "double"
                }
            ]
        }
    }
}

最佳实践

1. 任务设计

• 合理设计同步策略
• 使用增量同步减少数据量
• 添加数据验证
• 设置错误处理策略

2. 性能优化

• 优化通道数量
• 调整批量大小
• 使用连接池
• 监控系统资源

3. 数据质量

• 添加数据验证
• 清洗脏数据
• 记录数据质量报告
• 定期检查数据一致性

4. 安全管理

• 加密敏感数据
• 限制访问权限
• 审计操作日志
• 定期备份数据

5. 监控报警

• 监控任务执行状态
• 设置性能报警
• 记录错误日志
• 定期检查任务运行情况

相关资源

• DataX 官方文档
• DataX GitHub 仓库
• 虚谷数据库官方文档
• DataX 虚谷数据库插件

参考文档

详细配置信息请参考：references/datax-configuration.md