DolphinScheduler 集成 SeaTunnel 数据同步指南

概述

本文档详细介绍如何在 DolphinScheduler 中集成 Apache SeaTunnel，实现强大的数据同步和数据集成功能。

一、架构设计

1.1 部署方式选择

方案一：SeaTunnel 独立部署（推荐 ⭐）

┌─────────────────────┐         ┌─────────────────────┐
│ DolphinScheduler    │  SSH    │   SeaTunnel         │
│ - Master (120,121)  │ ──────► │ - Cluster           │
│ - Worker (121,122)  │  调用   │ - Standalone        │
└─────────────────────┘         └─────────────────────┘
                                        │
                                        ▼
                               ┌─────────────────────┐
                               │   数据源/目标       │
                               │ - MySQL/PostgreSQL  │
                               │ - Doris/StarRocks   │
                               │ - Kafka/HDFS        │
                               └─────────────────────┘

优势：

✅ 资源隔离：SeaTunnel 不占用 DS 集群资源
✅ 独立扩展：可根据数据量单独扩容 SeaTunnel
✅ 版本独立：DS 和 SeaTunnel 可以独立升级
✅ 故障隔离：SeaTunnel 故障不影响 DS 调度

适用场景：

大数据量同步任务
高并发数据集成
生产环境

方案二：SeaTunnel 本地部署

┌─────────────────────────────────────┐
│  DolphinScheduler Worker 节点       │
│  ├─ WorkerServer                    │
│  └─ SeaTunnel (Local Mode)         │
└─────────────────────────────────────┘

优势：

✅ 部署简单
✅ 调试方便

劣势：

❌ 资源争抢
❌ 扩展性差

适用场景：

小数据量同步
测试环境

1.2 网络规划

组件	IP/主机	端口	说明
DS Worker	192.168.192.121	-	提交任务
DS Worker	192.168.192.122	-	提交任务
SeaTunnel	192.168.192.130	8080	HTTP API（可选）
PostgreSQL	192.168.192.111	5432	数据源

二、SeaTunnel 独立部署方案

2.1 安装 SeaTunnel

在新的服务器（192.168.192.130）上执行：

# 创建 seaTunnel 用户
sudo useradd -m -s /bin/bash seatunnel

# 下载 SeaTunnel
cd /tmp
wget https://archive.apache.org/dist/seatunnel/2.3.8/apache-seatunnel-2.3.8-bin.tar.gz

# 解压
tar -xzf apache-seatunnel-2.3.8-bin.tar.gz
sudo mv apache-seatunnel-2.3.8 /opt/seatunnel
sudo chown -R seatunnel:seatunnel /opt/seatunnel

# 设置环境变量
su - seatunnel
echo 'export SEATUNNEL_HOME=/opt/seatunnel' >> ~/.bashrc
echo 'export PATH=$SEATUNNEL_HOME/bin:$PATH' >> ~/.bashrc
source ~/.bashrc

2.2 配置 SeaTunnel

cd /opt/seatunnel/config

# 复制示例配置
cp hazelcast.yaml.template hazelcast.yaml
cp log4j2.properties.template log4j2.properties

配置 `application.yaml`

seatunnel:
  # 引擎配置
  engine:
    history:
      max-history: 30
    
  # REST API 配置（可选）
  rest:
    enabled: true
    port: 8080
  
  # 检查点配置
  checkpoint:
    interval: 10000  # 10 秒
    timeout: 60000   # 60 秒

2.3 配置 Connector

根据需要同步的数据源，下载相应的 Connector：

cd /opt/seatunnel/connectors

# 示例：下载 MySQL Connector（通常已包含）
# 如果需要额外的 connector，从官方下载并放置到此目录

2.4 测试 SeaTunnel

创建测试任务配置文件 test_job.conf：

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    row.num = 10
    schema = {
      fields {
        name = string
        age = int
      }
    }
  }
}

sink {
  Console {
  }
}

执行测试：

su - seatunnel
$SEATUNNEL_HOME/bin/seatunnel.sh --config test_job.conf

三、DolphinScheduler 集成配置

3.1 在 DS 中预留 SeaTunnel 环境变量

在 DolphinScheduler 的 dolphinscheduler_env.sh 中添加：

# ===========================================
# SeaTunnel 集成配置
# ===========================================

# SeaTunnel 安装路径
export SEATUNNEL_HOME=/opt/seatunnel

# SeaTunnel 配置文件路径
export SEATUNNEL_CONF=$SEATUNNEL_HOME/config

# SeaTunnel Connector 路径
export SEATUNNEL_LIB=$SEATUNNEL_HOME/lib

# SeaTunnel 日志路径
export SEATUNNEL_LOG=$SEATUNNEL_HOME/logs

# Java 环境（SeaTunnel 需要）
export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64

注意： 如果 SeaTunnel 部署在其他服务器，此变量指向本地路径或留空，通过 SSH 远程调用。

3.2 配置 SSH 信任（跨服务器调用）

如果 SeaTunnel 在独立服务器（192.168.192.130），需要在 DS Worker 节点配置 SSH 信任：

# 在 DS Worker 节点执行（node-2 和 node-3）
su - dolphinscheduler
ssh-keygen -t rsa -P ""
ssh-copy-id seatunnel@192.168.192.130

# 测试连接
ssh seatunnel@192.168.192.130 "hostname"

3.3 创建 SeaTunnel 任务类型

在 DolphinScheduler Web UI 中：

方法一：使用 Shell 任务（推荐）

创建工作流定义
添加 Shell 任务节点
填写脚本内容：

#!/bin/bash

# 加载环境变量
source /opt/dolphinscheduler/conf/dolphinscheduler_env.sh

# SeaTunnel 任务配置
SEATUNNEL_JOB_CONFIG="/opt/seatunnel/jobs/mysql_to_doris.conf"

# 执行 SeaTunnel 任务
$SEATUNNEL_HOME/bin/seatunnel.sh \
  --config $SEATUNNEL_JOB_CONFIG \
  --deploy-mode client

# 检查执行结果
if [ $? -eq 0 ]; then
  echo "SeaTunnel 任务执行成功"
  exit 0
else
  echo "SeaTunnel 任务执行失败"
  exit 1
fi

方法二：使用 Python 任务

import subprocess
import sys
from datetime import datetime

def execute_seatunnel_job(job_config):
    """
    执行 SeaTunnel 任务
    
    Args:
        job_config: SeaTunnel 配置文件路径
        
    Returns:
        bool: 执行结果
    """
    try:
        # 构建命令
        cmd = f"""
        source /opt/dolphinscheduler/conf/dolphinscheduler_env.sh &&
        $SEATUNNEL_HOME/bin/seatunnel.sh --config {job_config}
        """
        
        # 执行命令
        result = subprocess.run(
            cmd,
            shell=True,
            capture_output=True,
            text=True,
            timeout=3600  # 超时时间：1 小时
        )
        
        # 输出日志
        print(f"=== SeaTunnel 任务开始时间：{datetime.now()} ===")
        print(result.stdout)
        
        if result.returncode != 0:
            print(f"错误：{result.stderr}")
            return False
            
        print(f"=== SeaTunnel 任务完成时间：{datetime.now()} ===")
        return True
        
    except subprocess.TimeoutExpired:
        print("错误：任务执行超时")
        return False
    except Exception as e:
        print(f"错误：{str(e)}")
        return False

# 主函数
if __name__ == "__main__":
    # 配置文件路径（可从参数传入）
    job_config = "${seatunnel_job_config}"
    
    success = execute_seatunnel_job(job_config)
    sys.exit(0 if success else 1)

四、实际应用场景

4.1 MySQL → Doris 数据同步

创建 SeaTunnel 配置文件 mysql_to_doris.conf：

env {
  parallelism = 2
  job.mode = "BATCH"
  checkpoint.interval = 10000
}

source {
  Jdbc {
    url = "jdbc:mysql://192.168.192.111:3306/source_db?useUnicode=true&characterEncoding=UTF-8"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "your_password"
    query = "SELECT * FROM users WHERE update_time > '${last_sync_time}'"
    result_table_name = "users_temp"
  }
}

transform {
  Filter {
    source_table_name = "users_temp"
    result_table_name = "users_filtered"
    condition = "age >= 18"
  }
}

sink {
  Doris {
    fenodes = "192.168.192.140:8030,192.168.192.141:8030"
    username = "root"
    password = "your_password"
    database = "target_db"
    table = "users"
    sink.label-prefix = "sync_users_${CURRENT_TIMESTAMP}"
    sink.enable-2pc = "false"
    sink.buffer-flush.max-rows = 100000
    sink.buffer-flush.max-bytes = 500000000
    sink.buffer-flush.interval-ms = 10000
  }
}

在 DolphinScheduler 中创建任务：

#!/bin/bash

# 设置上次同步时间（可从数据库或参数获取）
LAST_SYNC_TIME=$(date -d "1 hour ago" '+%Y-%m-%d %H:%M:%S')

# 替换配置文件中的变量
sed -i "s/\${last_sync_time}/$LAST_SYNC_TIME/g" \
  /opt/seatunnel/jobs/mysql_to_doris.conf

# 执行同步任务
/opt/seatunnel/bin/seatunnel.sh \
  --config /opt/seatunnel/jobs/mysql_to_doris.conf \
  --deploy-mode client

# 记录本次同步时间
echo $(date '+%Y-%m-%d %H:%M:%S') > /opt/seatunnel/state/last_sync_time

4.2 PostgreSQL → Kafka 实时同步

创建配置文件 pg_to_kafka.conf：

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  Postgres-CDC {
    base-url = "jdbc:postgresql://192.168.192.111:5432/source_db"
    username = "postgres"
    password = "your_password"
    table-names = ["public.orders", "public.order_items"]
    schema-names = ["public"]
  }
}

sink {
  Kafka {
    bootstrap.servers = "192.168.192.150:9092,192.168.192.151:9092"
    topic = "orders_topic"
    semantics = "EXACTLY_ONCE"
    format = "json"
  }
}

4.3 多表批量同步

创建批量同步脚本 batch_sync.sh：

#!/bin/bash

# 定义要同步的表
TABLES=("users" "orders" "order_items" "products")

# 循环执行每个表的同步
for TABLE in "${TABLES[@]}"; do
  echo "开始同步表：$TABLE"
  
  # 动态生成配置文件
  cat > /opt/seatunnel/jobs/sync_${TABLE}.conf << EOF
env {
  parallelism = 2
  job.mode = "BATCH"
}

source {
  Jdbc {
    url = "jdbc:mysql://192.168.192.111:3306/source_db"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "your_password"
    query = "SELECT * FROM $TABLE"
    result_table_name = "${TABLE}_temp"
  }
}

sink {
  Doris {
    fenodes = "192.168.192.140:8030"
    username = "root"
    password = "your_password"
    database = "target_db"
    table = "$TABLE"
    sink.label-prefix = "sync_${TABLE}_\${CURRENT_TIMESTAMP}"
  }
}
EOF

  # 执行同步
  /opt/seatunnel/bin/seatunnel.sh \
    --config /opt/seatunnel/jobs/sync_${TABLE}.conf \
    --deploy-mode client
  
  if [ $? -ne 0 ]; then
    echo "表 $TABLE 同步失败"
    exit 1
  fi
  
  echo "表 $TABLE 同步完成"
done

echo "所有表同步完成"

五、高级配置

5.1 资源限制

为防止 SeaTunnel 任务占用过多资源，在 DS Worker 上配置资源限制：

# 在 dolphinscheduler_worker_server.vmoptions 中
-Xmx2g  # 最大堆内存
-Xms1g  # 最小堆内存

或在 systemd 服务中限制：

# /etc/systemd/system/dolphinscheduler-worker.service
[Service]
...
MemoryLimit=4G
CPUQuota=200%

5.2 容错配置

在 SeaTunnel 配置中启用容错：

env {
  checkpoint.interval = 10000
  checkpoint.timeout = 60000
  restart.strategy = "fixed-delay"
  restart.times = 3
  restart.delay = 10000
}

5.3 监控告警

配置日志监控

# 创建监控脚本
cat > /opt/monitor/seatunnel_monitor.sh << 'EOF'
#!/bin/bash

LOG_FILE="/opt/seatunnel/logs/seatunnel-engine.log"
ERROR_COUNT=$(grep -c "ERROR" $LOG_FILE | tail -1)

if [ $ERROR_COUNT -gt 10 ]; then
  echo "SeaTunnel 任务错误数超过阈值：$ERROR_COUNT"
  # 发送告警（邮件、钉钉、企业微信等）
  # send_alert "SeaTunnel 错误告警" "错误数：$ERROR_COUNT"
fi
EOF

chmod +x /opt/monitor/seatunnel_monitor.sh

# 添加到 crontab
crontab -e
*/5 * * * * /opt/monitor/seatunnel_monitor.sh

六、DataX 集成方案

6.1 DataX 安装

# 下载 DataX
cd /tmp
wget https://github.com/alibaba/DataX/releases/download/v3.0.0/datax.tar.gz
tar -xzf datax.tar.gz
sudo mv datax /opt/datax
sudo chown -R dolphinscheduler:dolphinscheduler /opt/datax

# 配置环境变量（在 dolphinscheduler_env.sh 中）
export DATAX_HOME=/opt/datax
export DATAX_PYTHON=/usr/bin/python3

6.2 DataX 任务示例

在 DolphinScheduler 中创建 Shell 任务：

#!/bin/bash

# DataX 作业配置
JOB_CONFIG="/opt/datax/job/mysql_to_hive.json"

# 执行 DataX 任务
$DATAX_HOME/bin/datax.py $JOB_CONFIG

# 检查结果
if [ $? -eq 0 ]; then
  echo "DataX 任务执行成功"
  exit 0
else
  echo "DataX 任务执行失败"
  exit 1
fi

6.3 DataX 配置示例 (`mysql_to_hive.json`)

{
  "job": {
    "setting": {
      "speed": {
        "channel": 3
      },
      "errorLimit": {
        "record": 0,
        "percentage": 0.02
      }
    },
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "username": "root",
            "password": "password",
            "column": ["*"],
            "connection": [
              {
                "table": ["users"],
                "jdbcUrl": ["jdbc:mysql://192.168.192.111:3306/source_db"]
              }
            ]
          }
        },
        "writer": {
          "name": "hdfswriter",
          "parameter": {
            "defaultFS": "hdfs://192.168.192.160:9000",
            "fileType": "text",
            "path": "/user/hive/warehouse/users",
            "fileName": "users",
            "column": [
              {"name": "id", "type": "bigint"},
              {"name": "name", "type": "string"}
            ],
            "writeMode": "append",
            "fieldDelimiter": ","
          }
        }
      }
    ]
  }
}

七、最佳实践

7.1 性能优化

并行度设置

env {
  parallelism = 4  # 根据 CPU 核心数设置
}

批处理大小

sink {
  Doris {
    sink.buffer-flush.max-rows = 100000
    sink.buffer-flush.max-bytes = 500000000
  }
}

JDBC 优化

source {
  Jdbc {
    partition.column = "id"
    partition.num = "10"
    fetchSize = 1000
  }
}

7.2 故障排查

常见问题 1：内存溢出

# 增加 JVM 参数
export SEATUNNEL_OPTS="-Xmx4g -Xms2g"

常见问题 2：连接超时

source {
  Jdbc {
    connection.check.timeout = 60000
    connection.timeout = 30000
  }
}

7.3 安全建议

使用密钥管理密码
```
# 使用 Doris 的 Secret 功能
```
网络隔离
- 数据库只允许 DS 和 SeaTunnel 访问
- 使用内网通信
定期审计
- 检查任务执行日志
- 监控数据访问记录

八、附录

8.1 完整的环境变量配置

# /opt/dolphinscheduler/conf/dolphinscheduler_env.sh

# ========== 基础配置 ==========
export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
export HADOOP_HOME=/opt/hadoop
export HIVE_HOME=/opt/hive

# ========== SeaTunnel 配置 ==========
export SEATUNNEL_HOME=/opt/seatunnel
export SEATUNNEL_CONF=$SEATUNNEL_HOME/config
export SEATUNNEL_LIB=$SEATUNNEL_HOME/lib
export SEATUNNEL_LOG=$SEATUNNEL_HOME/logs
export SEATUNNEL_OPTS="-Xmx4g -Xms2g"

# ========== DataX 配置 ==========
export DATAX_HOME=/opt/datax
export DATAX_PYTHON=/usr/bin/python3

# ========== Spark 配置（可选） ==========
export SPARK_HOME=/opt/spark
export SPARK_MASTER=spark://192.168.192.170:7077

8.2 常用连接器列表

数据源	Connector	支持情况
MySQL	jdbc	✅ 内置
PostgreSQL	jdbc	✅ 内置
Doris/StarRocks	doris	✅ 内置
Oracle	jdbc	✅ 内置
SQL Server	jdbc	✅ 内置
Kafka	kafka	✅ 内置
HDFS	hdfs	✅ 内置
Hive	hive	✅ 内置
MongoDB	mongodb	✅ 内置
Elasticsearch	elasticsearch	✅ 内置

概述​

一、架构设计​

1.1 部署方式选择​

方案一：SeaTunnel 独立部署（推荐 ⭐）​

方案二：SeaTunnel 本地部署​

1.2 网络规划​

二、SeaTunnel 独立部署方案​

2.1 安装 SeaTunnel​

2.2 配置 SeaTunnel​

配置 application.yaml​

2.3 配置 Connector​

2.4 测试 SeaTunnel​

三、DolphinScheduler 集成配置​

3.1 在 DS 中预留 SeaTunnel 环境变量​

3.2 配置 SSH 信任（跨服务器调用）​

3.3 创建 SeaTunnel 任务类型​

方法一：使用 Shell 任务（推荐）​

方法二：使用 Python 任务​

四、实际应用场景​

4.1 MySQL → Doris 数据同步​

4.2 PostgreSQL → Kafka 实时同步​

4.3 多表批量同步​

五、高级配置​

5.1 资源限制​

5.2 容错配置​

5.3 监控告警​

配置日志监控​

六、DataX 集成方案​

6.1 DataX 安装​

6.2 DataX 任务示例​

6.3 DataX 配置示例 (mysql_to_hive.json)​

七、最佳实践​

7.1 性能优化​

7.2 故障排查​

常见问题 1：内存溢出​

常见问题 2：连接超时​

7.3 安全建议​

八、附录​

8.1 完整的环境变量配置​

8.2 常用连接器列表​

8.3 参考资源​

概述