seatunnel connector 连接器
source 连接器:读取原数据的连接器,支持数据源种类包括:关系型,非关系型,图形数据库等等57种数据库。
transform 连接器:转换器---将原数据源数据转换成目标源连接器的工具。
slink 连接器:写入目标数据源连接器,支持关系型,非关系型,图形数据库等等55种数据库
执行引擎选择
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>
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'),?,?)"
}
}