seatunnel engine 的master
服务和worker
服务分离,每个服务单独一个进程。
- master节点只负责作业调度,restful api,任务提交等,imap数据只存储在master节点中。
- worker节点只负责任务的执行,不参与选举成为master,也不存储imap数据。
在所有master
节点中,同一时间只有一个master
节点工作,其他master
节点处于standby
状态。
当master
节点宕机或心跳超时,会从其它节点中选举出一个新的master active
节点。
这是最推荐的一种使用方式,在该模式下master的负载会很小,master
有更多的资源用来进行作业的调度,任务的容错指标监控以及提供rest api
服务等,会有更高的稳定性。
同时worker
节点不存储imap
的数据,所有的imap
数据都存储在master
节点中,即使worker
节点负载高或者挂掉,也不会导致imap
数据重新分布。
下载安装包
在开始下载seatunnel之前,您需要确保您已经安装了seatunnel所需要的以下软件:
安装java (java 8 或 11, 其他高于java 8的版本理论上也可以工作) 以及设置
java_home
。
进入seatunnel下载页面(https://seatunnel.apache.org/download)下载最新版本的发布版安装包`seatunnel- -bin.tar.gz`
或者您也可以通过终端下载
export version="2.3.8" wget "https://archive.apache.org/dist/seatunnel/${version}/apache-seatunnel-${version}-bin.tar.gz" tar -xzvf "apache-seatunnel-${version}-bin.tar.gz"
配置 seatunnel_home
您可以通过添加 /etc/profile.d/seatunnel.sh
文件来配置 seatunnel_home
。
/etc/profile.d/seatunnel.sh
的内容如下:
export seatunnel_home=${seatunnel install path} export path=$path:$seatunnel_home/bin
配置 master 节点 jvm 选项
master节点的jvm参数在$seatunnel_home/config/jvm_master_options
文件中配置。
# jvm heap -xms2g -xmx2g # jvm dump -xx:+heapdumponoutofmemoryerror -xx:heapdumppath=/tmp/seatunnel/dump/zeta-server # metaspace -xx:maxmetaspacesize=2g # g1gc -xx:+useg1gc
worker节点的jvm参数在$seatunnel_home/config/jvm_worker_options
文件中配置。
# jvm heap -xms2g -xmx2g # jvm dump -xx:+heapdumponoutofmemoryerror -xx:heapdumppath=/tmp/seatunnel/dump/zeta-server # metaspace -xx:maxmetaspacesize=2g # g1gc -xx:+useg1gc
配置 seatunnel engine
seatunnel engine 提供许多功能,需要在 seatunnel.yaml
中进行配置。
imap中数据的备份数设置(该参数在worker节点无效)
seatunnel engine 基于 hazelcast imdg
实现集群管理。集群的状态数据(作业运行状态、资源状态)存储在 hazelcast imap
。
存储在 hazelcast imap
中的数据将在集群的所有节点上分布和存储。
hazelcast
会分区存储在 imap
中的数据。每个分区可以指定备份数量。因此,seatunnel engine 可以实现集群 ha,无需使用其他服务(例如 zookeeper)。
backup count
是定义同步备份数量的参数。例如,如果设置为 1,则分区的备份将放置在一个其他成员上。如果设置为 2,则将放置在两个其他成员上。
我们建议 backup-count
的值为 min(1, max(5, n/2))。n 是集群节点的数量。
seatunnel: engine: backup-count: 1 # 其他配置
由于在分离集群模式下,worker节点不存储imap数据,因此worker节点的
backup-count
配置无效。
如果master和worker进程在同一个机器上启动,master和worker会共用seatunnel.yaml
配置文件,此时worker节点服务会忽略backup-count
配置。
slot配置(该参数在master节点无效)
slot数量决定了集群节点可以并行运行的任务组数量。一个任务需要的slot的个数公式为 n = 2 + p(任务配置的并行度)。
默认情况下seatunnel engine的slot个数为动态,即不限制个数。我们建议slot的个数设置为节点cpu核心数的2倍。
动态slot个数(默认)配置如下:
seatunnel: engine: slot-service: dynamic-slot: true # 其他配置
静态slot个数配置如下:
seatunnel: engine: slot-service: dynamic-slot: false slot-num: 20
由于在分离集群模式下,master节点不运行任务,所以master服务不会启动slot服务,因此master节点的slot-service
配置无效。
如果master和worker进程在同一个机器上启动,master和worker会共用seatunnel.yaml
配置文件,此时master节点服务会忽略slot-service
配置。
检查点管理器(该参数在worker节点无效)
与 flink 一样,seatunnel engine 支持 chandy–lamport
算法。因此,可以实现无数据丢失和重复的数据同步。
interval
两个检查点之间的间隔,单位是毫秒。如果在作业配置文件的 env 中配置了 checkpoint.interval
参数,将以作业配置文件中设置的为准。
timeout
检查点的超时时间。如果在超时时间内无法完成检查点,则会触发检查点失败,作业失败。如果在作业的配置文件的env中配置了checkpoint.timeout
参数,将以作业配置文件中设置的为准。
示例
seatunnel: engine: backup-count: 1 print-execution-info-interval: 10 slot-service: dynamic-slot: true checkpoint: interval: 300000 timeout: 10000
checkpoint storage
检查点是一种容错恢复机制。这种机制确保程序在运行时,即使突然遇到异常,也能自行恢复。
检查点定时触发,每次检查点进行时每个task都会被要求将自身的状态信息(比如读取kafka时读取到了哪个offset)上报给检查点线程,由该线程写入一个分布式存储(或共享存储)。
当任务失败然后自动容错恢复时,或者通过seatunnel.sh -r
指令恢复之前被暂停的任务时,会从检查点存储中加载对应作业的状态信息,并基于这些状态信息进行作业的恢复。
如果集群的节点大于1,检查点存储必须是一个分布式存储,或者共享存储,这样才能保证任意节点挂掉后依然可以在另一个节点加载到存储中的任务状态信息。
检查点配置只有master服务才会读取,worker服务不会读取检查点配置。如果master和worker进程在同一个机器上启动,master和worker会共用
seatunnel.yaml
配置文件,此时worker节点服务会忽略checkpoint
配置。
历史作业过期配置
每个完成的作业的信息,如状态、计数器和错误日志,都存储在 imap
对象中。
随着运行作业数量的增加,内存会增加,最终内存将溢出。因此,您可以调整 history-job-expire-minutes
参数来解决这个问题。
此参数的时间单位是分钟。默认值是 1440 分钟,即一天。
示例
seatunnel: engine: history-job-expire-minutes: 1440
类加载器缓存模式
此配置主要解决不断创建和尝试销毁类加载器所导致的资源泄漏问题。如果您遇到与metaspace空间溢出相关的异常,您可以尝试启用此配置。
为了减少创建类加载器的频率,在启用此配置后,seatunnel 在作业完成时不会尝试释放相应的类加载器,以便它可以被后续作业使用,也就是说,当运行作业中使用的 source/sink
连接器类型不是太多时,它更有效。默认值是 false。
示例
seatunnel: engine: classloader-cache-mode: true
imap持久化配置(该参数在worker节点无效)
由于在分离集群模式下,只有master节点存储imap数据,worker节点不存储imap数据,所以worker服务不会读取该参数项。
在seatunnel中,我们使用imap(一种分布式的map,可以实现数据跨节点跨进程的写入的读取 有关详细信息,请参阅 hazelcast map
) 来存储每个任务及其task的状态,以便在任务所在节点宕机后,可以在其他节点上获取到任务之前的状态信息,从而恢复任务实现任务的容错。
默认情况下imap的信息只是存储在内存中,我们可以设置imap数据的复本数,具体可参考(imap中数据的备份数设置),如果复本数是2,代表每个数据会同时存储在2个不同的节点中。
一旦节点宕机,imap中的数据会重新在其它节点上自动补充到设置的复本数。但是当所有节点都被停止后,imap中的数据会丢失。当集群节点再次启动后,所有之前正在运行的任务都会被标记为失败,需要用户手工通过seatunnel.sh -r
指令恢复运行。
为了解决这个问题,我们可以将imap中的数据持久化到外部存储中,如hdfs、oss等。这样即使所有节点都被停止,imap中的数据也不会丢失,当集群节点再次启动后,所有之前正在运行的任务都会被自动恢复。
下面介绍如何使用 mapstore
持久化配置。
type
imap
持久化的类型,目前仅支持 hdfs
。
namespace
它用于区分不同业务的数据存储位置,如 oss 存储桶名称。
clustername
此参数主要用于集群隔离, 我们可以使用它来区分不同的集群,如 cluster1、cluster2,这也用于区分不同的业务。
fs.defaultfs
我们使用 hdfs api
读写文件,因此使用此存储需要提供 hdfs
配置。
如果您使用 hdfs,可以像这样配置:
map: engine*: map-store: enabled: true initial-mode: eager factory-class-name: org.apache.seatunnel.engine.server.persistence.filemapstorefactory properties: type: hdfs namespace: /tmp/seatunnel/imap clustername: seatunnel-cluster storage.type: hdfs fs.defaultfs: hdfs://localhost:9000
如果没有 hdfs,并且您的集群只有一个节点,您可以像这样配置使用本地文件:
map: engine*: map-store: enabled: true initial-mode: eager factory-class-name: org.apache.seatunnel.engine.server.persistence.filemapstorefactory properties: type: hdfs namespace: /tmp/seatunnel/imap clustername: seatunnel-cluster storage.type: hdfs fs.defaultfs: file:///
如果您使用 oss,可以像这样配置:
map: engine*: map-store: enabled: true initial-mode: eager factory-class-name: org.apache.seatunnel.engine.server.persistence.filemapstorefactory properties: type: hdfs namespace: /tmp/seatunnel/imap clustername: seatunnel-cluster storage.type: oss block.size: block size(bytes) oss.bucket: oss://bucket name/ fs.oss.accesskeyid: oss access key id fs.oss.accesskeysecret: oss access key secret fs.oss.endpoint: oss endpoint
注意:使用oss 时,确保 lib目录下有这几个jar.
aliyun-sdk-oss-3.13.2.jar hadoop-aliyun-3.3.6.jar jdom2-2.0.6.jar netty-buffer-4.1.89.final.jar netty-common-4.1.89.final.jar seatunnel-hadoop3-3.1.4-uber.jar
作业调度策略
当资源不足时,作业调度策略可以配置为以下两种模式:
wait:等待资源可用。
reject:拒绝作业,默认值。
示例
seatunnel: engine: job-schedule-strategy: wait
当dynamic-slot: ture
时,job-schedule-strategy: wait
配置会失效,将被强制修改为job-schedule-strategy: reject
,因为动态slot时该参数没有意义,可以直接提交。
配置网络服务
所有 seatunnel engine 网络相关的配置都在 hazelcast-master.yaml
和hazelcast-worker.yaml
文件中.
集群名称
seatunnel engine 节点使用 cluster-name
来确定另一个节点是否与自己在同一集群中。如果两个节点之间的集群名称不同,seatunnel 引擎将拒绝服务请求。
网络
基于 hazelcast
, 一个 seatunnel engine 集群是由运行 seatunnel engine 服务器的集群成员组成的网络。集群成员自动加入一起形成集群。这种自动加入是通过集群成员使用的各种发现机制来相互发现的。
请注意,集群形成后,集群成员之间的通信始终通过 tcp/ip 进行,无论使用的发现机制如何。
seatunnel engine 使用以下发现机制。
tcp
您可以将 seatunnel engine 配置为完整的 tcp/ip 集群。有关配置详细信息,请参阅 discovering members by tcp section
。
在分离集群模式下,master和worker服务使用不同的端口。
master节点网络配置 hazelcast-master.yaml
hazelcast: cluster-name: seatunnel network: rest-api: enabled: true endpoint-groups: cluster_write: enabled: true data: enabled: true join: tcp-ip: enabled: true member-list: - master-node-1:5801 - master-node-2:5801 - worker-node-1:5802 - worker-node-2:5802 port: auto-increment: false port: 5801 properties: hazelcast.heartbeat.failuredetector.type: phi-accrual hazelcast.heartbeat.interval.seconds: 2 hazelcast.max.no.heartbeat.seconds: 180 hazelcast.heartbeat.phiaccrual.failuredetector.threshold: 10 hazelcast.heartbeat.phiaccrual.failuredetector.sample.size: 200 hazelcast.heartbeat.phiaccrual.failuredetector.min.std.dev.millis: 100
worker节点网络配置 hazelcast-worker.yaml
hazelcast: cluster-name: seatunnel network: join: tcp-ip: enabled: true member-list: - master-node-1:5801 - master-node-2:5801 - worker-node-1:5802 - worker-node-2:5802 port: auto-increment: false port: 5802 properties: hazelcast.heartbeat.failuredetector.type: phi-accrual hazelcast.heartbeat.interval.seconds: 2 hazelcast.max.no.heartbeat.seconds: 180 hazelcast.heartbeat.phiaccrual.failuredetector.threshold: 10 hazelcast.heartbeat.phiaccrual.failuredetector.sample.size: 200 hazelcast.heartbeat.phiaccrual.failuredetector.min.std.dev.millis: 100
tcp 是我们建议在独立 seatunnel engine 集群中使用的方式。
启动 seatunnel engine master 节点
可以通过守护进程使用 -d 参数启动。
mkdir -p $seatunnel_home/logs ./bin/seatunnel-cluster.sh -d -r master
日志将写入 $seatunnel_home/logs/seatunnel-engine-master.log
启动 seatunnel engine worker 节点
可以通过守护进程使用 -d
参数启动。
mkdir -p $seatunnel_home/logs ./bin/seatunnel-cluster.sh -d -r worker
日志将写入 $seatunnel_home/logs/seatunnel-engine-worker.log
安装 seatunnel engine 客户端
您可以通过添加 /etc/profile.d/seatunnel.sh
文件来配置 seatunnel_home
。/etc/profile.d/seatunnel.sh
的内容如下:
export seatunnel_home=${seatunnel install path} export path=$path:$seatunnel_home/bin
提交作业和管理作业 使用 seatunnel engine 客户端提交作业
安装 seatunnel engine 客户端
设置和服务器一样的seatunnel_home
您可以通过添加 /etc/profile.d/seatunnel.sh
文件来配置 seatunnel_home
。
/etc/profile.d/seatunnel.sh
的内容如下:
export seatunnel_home=${seatunnel install path} export path=$path:$seatunnel_home/bin
配置 seatunnel engine 客户端
所有 seatunnel engine 客户端的配置都在 hazelcast-client.yaml
里。
cluster-name
客户端必须与 seatunnel engine 具有相同的 cluster-name
。
否则,seatunnel engine 将拒绝客户端的请求。
network
需要将所有 seatunnel engine master节点的地址添加到这里。
hazelcast-client: cluster-name: seatunnel properties: hazelcast.logging.type: log4j2 network: cluster-members: - master-node-1:5801 - master-node-2:5801
提交作业和管理作业
现在集群部署完成了,您可以通过以下教程完成作业的提交和管理:
bin/seatunnel.sh --config $seatunnel_home/config/v2.batch.config.template
--async参数可以让作业在后台运行,当作业提交后,客户端会退出。
./bin/seatunnel.sh --config $seatunnel_home/config/v2.batch.config.template --async
-n或--name参数可以指定作业的名称
./bin/seatunnel.sh --config $seatunnel_home/config/v2.batch.config.template --async -n myjob
查看作业列表
./bin/seatunnel.sh -l
该命令会输出所有当前集群中的作业列表(包含运行完成的历史作业和正在运行的作业)
查看作业状态
./bin/seatunnel.sh -j <jobid>
该命令会输出指定作业的状态信息
获取正在运行的作业监控信息
./bin/seatunnel.sh --get_running_job_metrics
该命令会输出正在运行的作业的监控信息
获取指定作业监控信息
--metrics 参数可以获取指定作业的监控信息
./bin/seatunnel.sh --metrics <jobid>
暂停作业
./bin/seatunnel.sh -s <jobid>
该命令会暂停指定作业,注意,只有开启了checkpoint
的作业才支持暂停作业(实时同步作业默认开启checkpoint,批处理作业默认不开启checkpoint需要通过在 env 中配置checkpoint.interval
来开启checkpoint)。
暂停作业是以split
为最小单位的,即暂停作业后,会等待当前正在运行的split
运行完成后再暂停。
任务恢复后,会从暂停的split
继续运行。
恢复作业
./bin/seatunnel.sh -r <jobid> -c $seatunnel_home/config/v2.batch.config.template
该命令会恢复指定作业,注意,只有开启了checkpoint的作业才支持恢复作业(实时同步作业默认开启checkpoint,批处理作业默认不开启checkpoint需要通过在 env 中配置checkpoint.interval来开启checkpoint)。
恢复作业需要指定jobid和作业的配置文件。
运行失败的作业和通过seatunnel.sh -s暂停的作业都可以通过该命令恢复。
取消作业
./bin/seatunnel.sh -can <jobid1> [<jobid2> <jobid3> ...]
该命令会取消指定作业,取消作业后,作业会被停止,作业的状态会变为canceled。
支持批量取消作业,可以一次取消多个作业。
被cancel的作业的所有断点信息都将被删除,无法通过seatunnel.sh -r恢复。
使用 rest api 提交作业
seatunnel engine 提供了 rest api
用于提交作业。
v2版本的api使用jetty
支持,与v1版本的接口规范相同 ,可以通过修改seatunnel.yaml
中的配置项来指定端口和context-path
,同时可以配置 enable-dynamic-port
开启动态端口(默认从 port 开始累加),默认为关闭, 如果enable-dynamic-port
为true,将使用port和port+port-range
范围内未使用的端口,默认范围是100。
seatunnel: engine: http: enable-http: true port: 8080 enable-dynamic-port: false port-range: 100
同时也可以配置context-path,配置如下:
seatunnel: engine: http: enable-http: true port: 8080 context-path: /seatunnel
其他api我已上传github请查看自取
【 github地址:https://github.com/mrkuhuo/data-warehouse-learning 】
【 gitee 地址:https://gitee.com/wzylzjtn/data-warehouse-learning 】
到此这篇关于apache seatunnel 集群部署详细教程的文章就介绍到这了,更多相关apache seatunnel 集群部署内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网!
发表评论