SeaTunnel部署及其Demo

SeaTunnel部署及其Demo

环境、安装下载（国内镜像下载）

• 环境准备：确保你的系统已安装 Java 8 或 11，并正确设置了 JAVA_HOME 环境变量。

• 下载，解压

https://mirrors.tuna.tsinghua.edu.cn/apache/seatunnel/2.3.12/apache-seatunnel-2.3.12-bin.tar.gztar -zxvf apache-seatunnel-2.3.12-bin.tar.gz

• 下载插件（根据需要下载）

安装连接器插件：从2.2.0-beta版本开始，二进制包默认不包含连接器，需要手动安装。进入解压后的SeaTunnel目录，执行安装脚本：如果需要指定连接器版本（例如2.3.8），则执行 sh bin/install-plugin.sh 2.3.8。你通常不需要全部连接器。可以编辑 config/plugin_config 文件，按格式（例如下方）指定所需插件。要让示例应用运行，通常需要 connector-fake 和 connector-console。连接器插件的作用
连接器插件 = 数据源驱动程序每个插件让 SeaTunnel 能够连接特定的数据源MySQL 插件：连接 MySQL 数据库Oracle 插件：连接 Oracle 数据库Console 插件：输出到控制台Fake 插件：生成测试数据

• 修改 config/plugin_config，只保留你需要的：（也可以不改，全部下载）

--connectors-v2--
connector-jdbc-mysql
connector-jdbc-oracle
connector-console  # 这个建议保留，用于调试输出
--end--

• 安装插件,执行命令

# 进入 SeaTunnel 目录
sh bin/install-plugin.sh

使用

• mysql官方文档
  https://seatunnel.apache.org/docs/2.3.3/connector-v2/source/Mysql
  https://seatunnel.apache.org/docs/2.3.3/connector-v2/sink/Mysql

• 使用mysql的话，需要拷贝jar包到seatunnel安装包/lib 下
  1、https://seatunnel.apache.org/docs/2.3.3/connector-v2/source/Mysql
  2、https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.33/mysql-connector-java-8.0.33.jar

• 增加测试数据

-- mysql建表
CREATE TABLE `t_8_100w`  (`id` bigint NOT NULL COMMENT '主键',`name` varchar(2000) NULL COMMENT '名字',`sex` int null COMMENT '性别：1男；2女',`decimal_f` decimal(32, 6) NULL COMMENT '大数字',`phone_number` varchar(20) DEFAULT '13456780000' COMMENT '电话',`age` varchar(255) NULL COMMENT '字符串年龄转数字',`create_time` timestamp DEFAULT CURRENT_TIMESTAMP COMMENT '新增时间',`description` longtext NULL COMMENT '大文本',`address` varchar(2000) NULL COMMENT '空地址转默认值：未知',PRIMARY KEY (`id`)
);-- 新增存储过程
DELIMITER $$
CREATE PROCEDURE InsertMultipleRows_Batch(IN start_id INT,        -- 起始IDIN end_id INT,          -- 结束IDIN batch_size INT       -- 批次大小
)
BEGINDECLARE i INT DEFAULT start_id;DECLARE description_text LONGTEXT;DECLARE address_text VARCHAR(255);DECLARE sex_text INT;DECLARE total_to_insert INT;SET total_to_insert = end_id - start_id;-- 开始事务START TRANSACTION;WHILE i < end_id DO-- 生成精确的1KB文本SET description_text = REPEAT(CONCAT('DataX_Test_Text_', i, '_ABCDEFGHIJKLMN_'), 41);-- 根据i%2生成地址IF i % 2 = 0 THENSET address_text = CONCAT('地址', i);SET sex_text = 1;ELSESET address_text = NULL;SET sex_text = 2;END IF;-- 插入数据INSERT INTO t_8_100w (`id`, `name`, `sex`, `decimal_f`, `age`, `description`, `address`) VALUES (i, CONCAT('名字', i), sex_text,i + 0.000001, ROUND((RAND() * 12) + 18),description_text,address_text);SET i = i + 1;-- 每batch_size条提交一次IF i % batch_size = 0 OR i = end_id THENCOMMIT;IF i < end_id THENSTART TRANSACTION;END IF;-- 显示进度IF i % 50000 = 0 OR i = end_id THENSELECT CONCAT('批次 ', start_id, '-', end_id, ': 已插入 ', i - start_id, ' / ', total_to_insert, ' 条记录') AS progress;END IF;END IF;END WHILE;SELECT CONCAT('批次完成! ID范围: ', start_id, ' 到 ', end_id - 1, ' (共', total_to_insert, '条)') AS batch_complete;
END$$
DELIMITER ;-- 分别执行新增数据
-- 测试1万条
CALL InsertMultipleRows_Batch(0, 10000, 500);
-- 每10万条创建一次，分批执行
CALL InsertMultipleRows_Batch(10000, 100000, 1000);
CALL InsertMultipleRows_Batch(100000, 200000, 1000);
CALL InsertMultipleRows_Batch(200000, 300000, 1000);
CALL InsertMultipleRows_Batch(300000, 400000, 1000);
CALL InsertMultipleRows_Batch(400000, 500000, 1000);
CALL InsertMultipleRows_Batch(500000, 600000, 1000);
CALL InsertMultipleRows_Batch(600000, 700000, 1000);
CALL InsertMultipleRows_Batch(700000, 800000, 1000);
CALL InsertMultipleRows_Batch(800000, 900000, 1000);
CALL InsertMultipleRows_Batch(900000, 1000000, 1000);

• 配置执行文件conf
  -- 2.3.x 新版本：必须使用 query，不再支持 table 参数
  

• DEMO1（直接把采集数据打印到控制面板）

# test2mysql.conf - 测试源数据
env {execution.parallelism = 2job.mode = "BATCH"
}source {Jdbc {url = "jdbc:mysql://ip:port/Cs1"driver = "com.mysql.cj.jdbc.Driver"user = "root"password = "******"query = "select * from t_sea_01"# 连接参数connection_check_timeout_sec = 300properties = {useUnicode = truecharacterEncoding = "utf8"serverTimezone = "Asia/Shanghai"}}
}sink {Console {}
}

执行命令

./data/seatunnel/apache-seatunnel-2.3.12/bin/seatunnel.sh --config ./data/seatunnel/myconf/test2mysql.conf -m local

查看结果

真背CPU啊（2核云服务器）

• DEMO2(mysql2mysql的不同库)

-- 在mysql另一个数据库执行
CREATE TABLE `t_8_100w_import`  (`id` bigint NOT NULL COMMENT '主键',`name` varchar(2000) NULL COMMENT '名字',`sex` int null COMMENT '性别：1男；2女',`decimal_f` decimal(32, 6) NULL COMMENT '大数字',`phone_number` varchar(20) COMMENT '电话',`age` varchar(255) NULL COMMENT '字符串年龄转数字',`create_time` timestamp COMMENT '新增时间',`description` longtext NULL COMMENT '大文本',`address` varchar(2000) NULL COMMENT '空地址转默认值：未知',PRIMARY KEY (`id`)
);

执行命令

./data/seatunnel/apache-seatunnel-2.3.12/bin/seatunnel.sh --config ./data/seatunnel/myconf/mysql2mysql.conf -m local

参考

https://www.jb51.net/program/2850931un.htm
https://www.cnblogs.com/robots2/p/17939863