seatunnel connector 连接器
source 连接器：读取原数据的连接器，支持数据源种类包括：关系型，非关系型，图形数据库等等57种数据库。
transform 连接器：转换器---将原数据源数据转换成目标源连接器的工具。
slink 连接器：写入目标数据源连接器，支持关系型，非关系型，图形数据库等等55种数据库
image.png
image.png
image.png
image.png
image.png
执行引擎选择
SeaTunnel Engine：
这是SeaTunnel自带的执行引擎，用于数据同步。它提供了高效且稳定的数据处理能力。
Apache Spark：
Spark是一个快速、通用的大规模数据处理引擎。SeaTunnel可以与Spark紧密集成，利用Spark的分布式计算能力来处理数据流。
Apache Flink：
Flink是一个流处理框架，用于无边界和有边界数据流的处理。SeaTunnel支持Flink引擎，允许用户构建实时数据流处理应用程序。
部署
（略）
注意多节点情况下本地文件读取时需要做策略处理，不然会存在部分时候报找不到文件错误。
查看部署引擎
# 假设你已经将 SeaTunnel 的 bin 目录添加到了 PATH 环境变量中
# 你可以使用 --help 或 -h 参数来获取帮助信息
# 对于基于 Spark 的 SeaTunnel
./start-waterdrop-spark.sh --help
# 或者对于其他可能的启动脚本
./start-seatunnel.sh --help
# 或者查看某个具体配置文件的帮助（如果支持的话）
./start-waterdrop-spark.sh -c config/your-config.conf --help
执行文件样例
env {
# You can set SeaTunnel environment configuration here
execution.parallelism = 10
job.mode = "BATCH"
checkpoint.interval = 10000
}
source {
jdbc {
result_table_name = "fake1"
url = "jdbc:oracle:thin:@10.1.0.10:1521:ora11g"
driver = "oracle.jdbc.OracleDriver"
user = "name"
password = "pwd"
batchSize = 1000
query = "select * from ALARM_LOG"
}
}
transform {
}
sink {
Console {
#控制台打印表数据
source_table_name = "fake1"
}
}
四部分：
- env：在SeaTunnel中，env通常用于设置运行环境的相关参数或配置。
- source：用于定义SeaTunnel需要在何处获取数据，并使用获取的数据进行下一步操作。你可以同时定义多个源，并且每个源都有自己的特定参数来定义如何获取数据。SeaTunnel会提取每个源将使用的参数，如result_table_name参数，用于指定当前源生成的数据名称，方便其他模块后续使用
- transform：当有了数据源后，业务可能需要进一步处理数据，因此SeaTunnel提供了转换模块transform。在SeaTunnel中，transform只能用于对数据进行一些简单的转换，例如将一列的数据转换为大写或小写，更改列名，或者将一列拆分为多列。同时，transform作为一个中间处理模块，可以同时使用result_table_name和source_table_name进行配置。
- sink:sink用于定义SeaTunnel需要把数据写到哪里。当定义多个源、转换时，每个源和转换读取哪些数据，需要使用result_table_name和source_table_name两个键进行配置。每个源端将配置一个result_table_name，以指示生成的数据源的名称，其他转换和接收模块可以使用source_table_name来引用对应的数据源名称。
其中源数据写法：
jdbc连接本地文件:
LocalFile{
result_table_name = "fake1"
delimiter = "\\|"
ignore_first_line = false
path = "/tmp/seatunnel/alarmlog.txt"
file_format_type = "text"
datatime_format = "yyyy-MM-dd HH:mm:ss"
schema {
fields {
ID = string
MSG= string
}
}
}
jdbc连接oracle:
jdbc {
url = "jdbc:oracle:thin:@10.1.0.10:1521:ora11g"
driver = "oracle.jdbc.OracleDriver"
user = "name"
password = "pwd"
batchSize = 1000
query = "select * from ALARM_LOG"
}
}
jdbc连接mysql:
jdbc {
url = "jdbc:mysql://10.1.1.1:3306/demo?useSSL=false&useUnicode=true&characterEncoding=utf-8"
driver = "com.mysql.cj.jdbc.Driver"
user = "root"
password = "root"
batchSize = 1000
result_table_name = "fake1"
query = "select * from d_sesl"
}
其中目标数据写法（sink）：
#打印控制台
Console {
}
#下载到数据库
jdbc {
url = "jdbc:mysql://10.1.1.10:3306/demo?useSSL=false&useUnicode=true&characterEncoding=utf-8"
driver = "com.mysql.cj.jdbc.Driver"
user = "root"
password = "1234"
batchSize = 1000
writeMode = "insert" # 或者使用 "update" 根据需要
fieldNames = ["id","je","se"] # 对应目标表的字段名
query = "insert into d_sesl(id,je,se) values(?,?,?)"
}
启动执行文件
查看seatunnel执行情况
#ps aux 和ps -ef
ps aux | grep seatunnel
启动执行文件并查看结果
# bin 目录下，命令
#全日志 执行本地节点？
./bin/seatunnel.sh --config <your-config-file.conf> -e local
#简日志
./bin/seatunnel.sh --config <your-config-file.conf>
image.png
seatunnel将本地文件读取存入oracle数据库完整样例
env {
execution.parallelism = 10
job.mode = "BATCH"
checkpoint.interval = 10000
}
source{
LocalFile{
result_table_name = "fake1"
delimiter = "\\|"
ignore_first_line = false
path = "/tmp/seatunnel/alarmlog.txt"
file_format_type = "text"
datatime_format = "yyyy-MM-dd HH:mm:ss"
schema {
fields {
ID = string
SERIALNUM = string
CLEARTIME=string
SENDMSG= string
MSG= string
}
}
}
}
transform {
}
sink {
jdbc {
url = "jdbc:oracle:thin:@10.10.0.10:1521:ora11g"
driver = "oracle.jdbc.OracleDriver"
user = "use"
password = "pwd"
batchSize = 1000
writeMode = "insert"
fieldNames = ["ID","SERIALNUM","CREATETIME","SENDMSG","MSG"] # 对应目标表的字段名
query = "insert into ALARM_LOG(ID,SERIALNUM,CREATETIME,SENDMSG,MSG) values(?,?,TO_DATE(?,'YYYY-MM-DD HH24:MI:SS'),?,?)"
}
}
网友评论