SeaTunnel MySQL users表到Doris同步操作指南

概述

本指南专门针对 MySQL → Canal → Kafka → SeaTunnel → Doris 的完整数据同步链路，提供从Kafka消费数据并写入Doris的配置和操作步骤。

数据流转路径：

MySQL big_data.users 表产生数据变更
Canal捕获binlog变更并发送到Kafka big_data_topic
SeaTunnel从Kafka消费数据并处理
SeaTunnel将处理后的数据写入Doris ods_db.users 表

重点聚焦于SeaTunnel从Kafka消费到Doris写入的配置文件编写和实际操作。

环境信息

数据库环境

数据源：MySQL big_data.users 表（通过Canal捕获变更）
中间层：Kafka big_data_topic（Canal同步主题）
目标库：Doris ods_db.users 表
同步工具：SeaTunnel 2.3.8

表结构映射

MySQL源表结构 (big_data.users)

CREATE TABLE `big_data`.`users` (
  `id` int NOT NULL AUTO_INCREMENT,
  `username` varchar(50) NOT NULL,
  `email` varchar(100) DEFAULT NULL,
  `age` int DEFAULT NULL,
  `gender` enum('M','F','OTHER') DEFAULT NULL,
  `created_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
  `updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `status` tinyint DEFAULT '1' COMMENT '1: active, 0: inactive',
  `salary` decimal(10,2) DEFAULT NULL,
  `city` varchar(50) DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `email` (`email`)
) ENGINE=InnoDB AUTO_INCREMENT=1018 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

Doris目标表结构 (ods_db.users)

CREATE TABLE `ods_db`.`users` (
  `id` int NOT NULL,
  `username` varchar(200) NOT NULL,
  `email` varchar(400) NULL,
  `age` int NULL,
  `gender` char(20) NULL,
  `created_at` datetime NULL,
  `updated_at` datetime NULL,
  `status` tinyint NULL COMMENT "1: active, 0: inactive",
  `salary` decimal(10,2) NULL,
  `city` varchar(200) NULL
) ENGINE=OLAP
UNIQUE KEY(`id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 10
PROPERTIES (
"replication_allocation" = "tag.location.default: 1",
"enable_unique_key_merge_on_write" = "true",
"light_schema_change" = "true"
);

基础环境

Kafka集群：192.168.192.211:9092
Doris集群：FE节点 192.168.192.141:9030
SeaTunnel：2.3.8版本（已部署）
数据主题：big_data_topic（参考mysql-ha-canal-kafka-integration-guide.md）

端口规划

组件	端口	用途
Kafka	9092	消息服务端口
Doris FE	9030	MySQL客户端连接端口
Doris FE	8030	Web监控页面
Doris BE	8040	BE监控页面

一、准备工作

1.1 确认数据链路状态

# 1. 检查MySQL源表数据和Canal同步状态
mysql -h<mysql_host> -u<user> -p -e "
SELECT 
    COUNT(*) as total_records,
    COUNT(DISTINCT email) as unique_emails,
    MIN(created_at) as earliest_record,
    MAX(created_at) as latest_record
FROM big_data.users;
"

# 2. 验证Canal是否正常捕获MySQL变更并发送到Kafka
kafka-console-consumer.sh --bootstrap-server 192.168.192.211:9092 --topic big_data_topic --from-beginning --max-messages 5

# 3. 检查Kafka主题状态
kafka-topics.sh --bootstrap-server 192.168.192.211:9092 --describe --topic big_data_topic

1.2 准备测试数据并验证链路

# 1. 创建测试数据插入脚本（通过MySQL插入验证完整链路）
cat > /tmp/users_test_data.sql << 'EOF'
USE big_data;

INSERT INTO users (username, email, age, gender, status, salary, city) VALUES
('张三', 'zhangsan@example.com', 28, 'M', 1, 8500.00, '北京'),
('李四', 'lisi@example.com', 32, 'F', 1, 9200.50, '上海'),
('王五', 'wangwu@example.com', 25, 'OTHER', 1, 7800.75, '广州'),
('赵六', 'zhaoliu@example.com', 35, 'M', 0, 12000.00, '深圳'),
('钱七', 'qianqi@example.com', 29, 'F', 1, 8800.25, '杭州');

-- 验证MySQL插入结果
SELECT * FROM users WHERE email IN ('zhangsan@example.com', 'lisi@example.com', 'wangwu@example.com', 'zhaoliu@example.com', 'qianqi@example.com');
EOF

# 2. 执行测试数据插入到MySQL
mysql -h<mysql_host> -u<user> -p < /tmp/users_test_data.sql

# 3. 验证Canal是否已将变更同步到Kafka（完整链路验证）
echo "等待Canal同步...约30秒后检查Kafka消息"
sleep 30
kafka-console-consumer.sh --bootstrap-server 192.168.192.211:9092 --topic big_data_topic --from-beginning --max-messages 10 --timeout-ms 10000

二、Doris目标表确认

2.1 验证目标表结构

-- 连接Doris验证目标表
mysql -h192.168.192.141 -P9030 -uroot -e "
USE ods_db;
DESC users;
SHOW CREATE TABLE users;
"

-- 检查表状态
mysql -h192.168.192.141 -P9030 -uroot -e "
SELECT 
    TABLE_NAME,
    TABLE_ROWS,
    DATA_LENGTH,
    INDEX_LENGTH
FROM information_schema.tables 
WHERE TABLE_SCHEMA = 'ods_db' AND TABLE_NAME = 'users';
"

三、SeaTunnel配置文件编写

3.1 创建配置目录

# 创建users同步专用配置目录
mkdir -p /opt/seatunnel/config/mysql-users-sync

# 进入配置目录
cd /opt/seatunnel/config/mysql-users-sync

3.2 主配置文件

创建 users-sync.conf 配置文件：

env {
  # 流处理模式
  execution.parallelism = 1
  job.mode = "STREAMING"
  # 检查点配置
  checkpoint.interval = 10000
  # 作业名称
  job.name = "MySQLUsersToDorisSync"
}

source {
  Kafka {
    # 从Kafka消费Canal同步的数据
    bootstrap.servers = "192.168.192.211:9092,192.168.192.212:9092,192.168.192.213:9092"
    # Canal同步的主题名
    topic = "big_data_topic"
    # 消费者组
    consumer.group = "seatunnel-users-consumer"
    # 从最早位置开始消费
    startup.mode = "earliest"
    # 数据格式（Canal输出的JSON格式）
    format = "json"
    
    # 精确的Schema定义（匹配users表结构）
    schema = {
      fields {
        id = "INT"
        username = "STRING"
        email = "STRING"
        age = "INT"
        gender = "STRING"  # enum类型在JSON中为字符串
        created_at = "TIMESTAMP"
        updated_at = "TIMESTAMP"
        status = "TINYINT"
        salary = "DECIMAL(10,2)"
        city = "STRING"
      }
    }
  }
}

sink {
  Doris {
    # Doris FE节点配置
    fenodes = "192.168.192.141:8030,192.168.192.142:8030,192.168.192.143:8030"
    # 目标数据库和表
    database = "ods_db"
    table = "users"
    # 认证信息
    username = "root"
    password = ""
    sink.label-prefix = "sync_${job_id}"
    sink.enable-2pc = "true"
    sink.enable-delete = "true"
    doris.config {
      format = "json"
      read_json_by_line = "true"
    }
  }
}

3.3 高级配置选项

如果需要更精细的控制，可以使用以下增强配置：

source {
  Kafka {
    # 消息确认机制
    enable.commit.on.checkpoint = true
    
    # 消费者高级配置
    kafka.config = {
      client.id = "seatunnel-employee-consumer"
      max.poll.records = 500
      session.timeout.ms = 45000
      heartbeat.interval.ms = 3000
      max.partition.fetch.bytes = 1048576
    }
    
    # 消费位置控制
    # startup.mode = "specific_offsets"
    # specific.offsets = {
    #   topic = "big_data_topic"
    #   partition = 0
    #   offset = 0
    # }
  }
}

sink {
  Doris {
    # 性能调优配置
    batch_max_rows = 50000
    batch_max_bytes = 10485760
    batch_interval_ms = 500
    
    # 连接池配置
    doris.config = {
      timeout = 600000
      max_conn = 20
      http.headers = {
        "Expect" = "100-continue"
      }
    }
    
    # 错误处理配置
    retry.configuration = {
      retry.times = 5
      retry.interval = 2000
      retry.enable = true
    }
  }
}

四、运行和验证

4.1 本地测试运行

# 进入SeaTunnel安装目录
cd /opt/bigdata/seatunnel-2.3.8/backend/apache-seatunnel-2.3.8

# 本地模式运行测试
./bin/seatunnel.sh --config /opt/seatunnel/config/mysql-users-sync/users-sync.conf -e local

# 监控运行日志
tail -f logs/seatunnel-engine-server.log | grep -E "users|ERROR|WARN"

4.2 集群模式部署

# 提交作业到SeaTunnel集群
curl -X POST http://localhost:8081/submit-job \
  -H "Content-Type: application/json" \
  -d '{
    "jobName": "MySQLUsersToDorisSync",
    "config": "'"$(cat /opt/seatunnel/config/mysql-users-sync/users-sync.conf | sed ':a;N;$!ba;s/\n/\\n/g')"'"
  }'

# 查询作业状态
curl http://localhost:8081/running-jobs

# 获取作业详细信息
curl http://localhost:8081/job-status/<job_id>

4.3 数据验证

# 1. 在Doris中验证数据同步结果
mysql -h192.168.192.141 -P9030 -uroot -e "
USE ods_db;
SELECT 
    COUNT(*) as total_records,
    COUNT(DISTINCT email) as unique_users,
    MIN(created_at) as earliest_record,
    MAX(created_at) as latest_record,
    AVG(salary) as avg_salary,
    COUNT(CASE WHEN status = 1 THEN 1 END) as active_users,
    COUNT(CASE WHEN status = 0 THEN 1 END) as inactive_users
FROM users;

-- 按城市统计
SELECT 
    city,
    COUNT(*) as user_count,
    AVG(salary) as avg_salary,
    COUNT(CASE WHEN gender = 'MALE' THEN 1 END) as male_count,
    COUNT(CASE WHEN gender = 'FEMALE' THEN 1 END) as female_count
FROM users 
WHERE city IS NOT NULL
GROUP BY city
ORDER BY user_count DESC;
"

# 2. 验证数据完整性
mysql -h192.168.192.141 -P9030 -uroot -e "
USE ods_db;
-- 检查是否有重复的email（应该没有，因为是UNIQUE KEY）
SELECT email, COUNT(*) as cnt 
FROM users 
WHERE email IS NOT NULL 
GROUP BY email 
HAVING cnt > 1;

-- 检查NULL值分布
SELECT 
    SUM(CASE WHEN username IS NULL THEN 1 ELSE 0 END) as null_username,
    SUM(CASE WHEN email IS NULL THEN 1 ELSE 0 END) as null_email,
    SUM(CASE WHEN age IS NULL THEN 1 ELSE 0 END) as null_age,
    SUM(CASE WHEN gender IS NULL THEN 1 ELSE 0 END) as null_gender
FROM users;
"

五、性能调优

5.1 并行度调优

env {
  execution.parallelism = 4  # 根据数据量和集群资源调整
  job.mode = "STREAMING"
}

source {
  Kafka {
    parallelism = 4  # 与topic分区数匹配
  }
}

5.2 批次大小调优

sink {
  Doris {
    batch_max_rows = 100000  # 增大批次大小提高吞吐量
    batch_interval_ms = 2000  # 延长批次间隔
    batch_max_bytes = 52428800  # 50MB批次大小限制
  }
}

5.3 内存和JVM调优

# 在启动脚本中添加JVM参数
export JAVA_OPTS="-Xms4g -Xmx8g -XX:+UseG1GC -XX:MaxGCPauseMillis=200"

# 或者在seatunnel-env.sh中配置
echo 'export JAVA_OPTS="-Xms4g -Xmx8g"' >> /opt/seatunnel/config/seatunnel-env.sh

六、故障排除

6.1 常见问题诊断

# 1. 检查Kafka连接
kafka-topics.sh --bootstrap-server 192.168.192.211:9092 --describe --topic big_data_topic

# 2. 检查Doris连接
mysql -h192.168.192.141 -P9030 -uroot -e "SHOW DATABASES;"

# 3. 检查SeaTunnel日志
tail -f /opt/bigdata/seatunnel-2.3.8/backend/apache-seatunnel-2.3.8/logs/seatunnel-engine-server.log | grep -E "ERROR|WARN"

# 4. 监控作业状态
curl http://localhost:8081/job-status/作业ID

6.2 数据一致性检查

-- 在Doris中验证数据完整性
mysql -h192.168.192.141 -P9030 -uroot -e "
SELECT 
    '数据总量' as metric,
    COUNT(*) as value
FROM ods_db.users
UNION ALL
SELECT 
    '活跃用户数',
    COUNT(*) 
FROM ods_db.users 
WHERE status = 1
UNION ALL
SELECT 
    '唯一邮箱数',
    COUNT(DISTINCT email)
FROM ods_db.users
WHERE email IS NOT NULL;
"

6.3 性能监控

# 1. 监控Kafka消费进度
kafka-consumer-groups.sh --bootstrap-server 192.168.192.211:9092 \
  --describe --group seatunnel-employee-consumer

# 2. 监控Doris导入状态
mysql -h192.168.192.141 -P9030 -uroot -e "
SELECT 
    table_name,
    state,
    timeout,
    load_start_time,
    load_end_time
FROM information_schema.loads 
WHERE db_name = 'ods_db'
ORDER BY load_start_time DESC 
LIMIT 10;
"

# 3. 监控SeaTunnel作业指标
curl http://localhost:8081/job-metrics/作业ID

七、运维管理

7.1 作业生命周期管理

#!/bin/bash
# mysql-users-sync-manager.sh

CONFIG_FILE="/opt/seatunnel/config/mysql-users-sync/users-sync.conf"
JOB_NAME="MySQLUsersToDorisSync"

case "$1" in
    start)
        echo "启动MySQL users表到Doris同步作业..."
        curl -X POST http://localhost:8081/submit-job \
          -H "Content-Type: application/json" \
          -d "{\"jobName\": \"$JOB_NAME\", \"config\": \"$(cat $CONFIG_FILE | sed ':a;N;$!ba;s/\n/\\n/g')\"}"
        ;;
    stop)
        echo "停止MySQL users表到Doris同步作业..."
        JOB_ID=$(curl -s http://localhost:8081/running-jobs | grep -o '"jobId":"[^"]*"' | head -1 | cut -d'"' -f4)
        if [ ! -z "$JOB_ID" ]; then
            curl -X POST http://localhost:8081/stop-job \
              -H "Content-Type: application/json" \
              -d "{\"jobId\": \"$JOB_ID\",\"isStopWithSavePoint\": true}"
        fi
        ;;
    status)
        echo "检查作业状态..."
        curl http://localhost:8081/running-jobs
        ;;
    *)
        echo "用法: $0 {start|stop|status}"
        exit 1
        ;;
esac

7.2 数据质量监控脚本

#!/bin/bash
# data-quality-monitor.sh

# 监控数据同步质量
LOG_FILE="/var/log/seatunnel/data-quality.log"

# 检查数据延迟
check_data_latency() {
    local latest_kafka_time=$(date -d "$(kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.192.211:9092 --topic big_data_topic --time -1 | cut -f3)" +%s)
    local latest_doris_time=$(mysql -h192.168.192.141 -P9030 -uroot -N -s -e "SELECT UNIX_TIMESTAMP(MAX(created_at)) FROM ods_db.users")
    
    local latency=$((latest_kafka_time - latest_doris_time))
    
    if [ $latency -gt 300 ]; then
        echo "$(date): WARNING - Data latency detected: ${latency} seconds" >> $LOG_FILE
        # 发送告警邮件
        echo "Data sync latency alert: ${latency} seconds" | mail -s "Data Quality Alert" admin@company.com
    fi
}

# 检查数据完整性
check_data_integrity() {
    local kafka_count=$(kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.192.211:9092 --topic big_data_topic --time -1 | awk -F: '{sum+=$3} END {print sum}')
    local doris_count=$(mysql -h192.168.192.141 -P9030 -uroot -N -s -e "SELECT COUNT(*) FROM ods_db.users")
    
    local diff=$((kafka_count - doris_count))
    local percentage=$(echo "scale=2; $diff*100/$kafka_count" | bc)
    
    if (( $(echo "$percentage > 1.0" | bc -l) )); then
        echo "$(date): WARNING - Data integrity issue: ${diff} records missing (${percentage}%)" >> $LOG_FILE
    fi
}

# 执行检查
check_data_latency
check_data_integrity

八、最佳实践

8.1 配置管理建议

版本控制：将配置文件纳入Git管理
环境分离：为开发、测试、生产环境维护不同的配置文件
参数模板化：使用变量替换关键配置参数
定期审查：每月审查和优化配置参数

8.2 性能优化建议

批次调优：根据数据量调整batch_max_rows和batch_interval_ms
并行度设置：source并行度应与Kafka topic分区数匹配
内存管理：合理设置JVM堆内存大小
网络优化：确保各组件间网络延迟最小化

8.3 监控告警建议

数据延迟监控：设置5分钟数据延迟告警阈值
作业状态监控：实时监控作业运行状态
资源使用监控：监控CPU、内存、磁盘使用率
数据质量监控：定期验证数据完整性和准确性

这个操作指南提供了从配置到运维的完整流程，可以直接用于生产环境的数据同步任务。

概述​

环境信息​

数据库环境​

表结构映射​

MySQL源表结构 (big_data.users)​

Doris目标表结构 (ods_db.users)​

基础环境​

端口规划​

一、准备工作​

1.1 确认数据链路状态​

1.2 准备测试数据并验证链路​

二、Doris目标表确认​

2.1 验证目标表结构​

三、SeaTunnel配置文件编写​

3.1 创建配置目录​

3.2 主配置文件​

3.3 高级配置选项​

四、运行和验证​

4.1 本地测试运行​

4.2 集群模式部署​

4.3 数据验证​

五、性能调优​

5.1 并行度调优​

5.2 批次大小调优​

5.3 内存和JVM调优​

六、故障排除​

6.1 常见问题诊断​

6.2 数据一致性检查​

6.3 性能监控​

七、运维管理​

7.1 作业生命周期管理​

7.2 数据质量监控脚本​

八、最佳实践​

8.1 配置管理建议​

8.2 性能优化建议​

8.3 监控告警建议​

概述