热门搜索:
您现在的位置是：首页 > 文章详情

万字长文 | Apache SeaTunnel 分离集群模式部署 K8s 集群实践

日期：2025-04-22点击：13

文章作者：雷宝鑫

整理排版：白鲸开源 曾辉 > Apache SeaTunnel官网链接: https://seatunnel.apache.org/

Apache SeaTunnel(以下简称SeaTunnel）是一款新一代高性能、分布式的数据集成同步工具，正受到业界广泛关注和应用。SeaTunnel支持三种部署模式：本地模式（Local）、混合集群模式（Hybrid Cluster Mode）和分离集群模式（Separated Cluster Mode）。

本文尝试介绍如何在K8s上以分离集群模式部署SeaTunnel，为有相关需求的伙伴提供完整的部署流程和配置案例参考。

前期准备

在开始部署之前，需要确保以下环境和组件已经准备就绪：

  • Kubernetes集群环境
  • kubectl命令行工具
  • docker
  • helm (option)

对于熟悉和具有Helm环境的部署，可以直接参考官网中使用Helm部署教程：

本文主要介绍基于Kubernetes环境和kubectl工具的方式实现部署。

构建SeaTunnel Docker镜像

目前官方已提供各版本的Docker镜像，可直接拉取，详细信息可参考官方文档：Set Up With Docker。

docker pull apache/seatunnel:<version_tag>

由于我们需要部署的是集群模式，接下来需要配置集群间的网络通信。SeaTunnel集群的网络服务是通过Hazelcast实现的，所以接下来对这部分内容进行配置。

Hazelcast集群相关配置

Headless Service配置

Hazelcast 集群是由运行 Hazelcast 的集群成员组成的网络，集群成员自动联合起来形成一个集群，这种自动加入是通过集群成员用于查找彼此的各种发现机制实现的。

Hazelcast 支持以下发现机制：

  • 自动发现机制，支持以下环境：
    • AWS
    • Azure
    • GCP
    • Kubernetes
  • TCP
  • Multicast
  • Eureka
  • Zookeeper

在本文的集群部署中，我们基于Hazelcast的Kubernetes自动发现机制来配置文件，详细的原理可以参考官网文档：Kubernetes Auto Discovery。

Hazelcast的k8s自动发现机制（DNS Lookup mode）需要借助于k8s的Headless Service功能来实现。

Headless Service在查询服务域名时，会将域名解析为所有匹配Pod的IP地址列表，以此来实现Hazelcast集群成员互相发现彼此。

为此，首先我们创建K8s Headless Service服务：

# use for hazelcast cluster join apiVersion: v1 kind: Service metadata: name: seatunnel-cluster spec: type: ClusterIP clusterIP: None selector: app.kubernetes.io/instance: seatunnel-cluster-app app.kubernetes.io/version: 2.3.10 ports: - port: 5801 name: hazelcast

上述配置中的关键部分：

  • metadata.name: seatunnel-cluster： 服务名称，Hazelcast 客户端/节点将通过该名称发现集群
  • spec.clusterIP: None：关键配置，声明为 Headless Service，不分配虚拟 IP
  • spec.selector: 选择器匹配的 Pod 标签，包含相应标签的pod会被该Service识别和代理
  • spec.port：Hazelcast的暴露端口

同时，为了能从系统外部利用rest api访问集群，我们定义另一个Service来包含Master的节点pod：

# use for access seatunnel from outside system via rest api apiVersion: v1 kind: Service metadata: name: seatunnel-cluster-master spec: type: ClusterIP clusterIP: None selector: app.kubernetes.io/instance: seatunnel-cluster-app app.kubernetes.io/version: 2.3.10 app.kubernetes.io/name: seatunnel-cluster-master app.kubernetes.io/component: master ports: - port: 8080 name: "master-port" targetPort: 8080 protocol: TCP

定义好上述K8s的Service服务后，接下来根据Hazelcast的k8s发现机制来配置hazelcast-master.yaml和hazelcast-worker.yaml文件。

Hazelcast master和worker的yaml配置

对于SeaTunnel分离集群模式来说，所有网络相关的配置都在hazelcast-master.yaml和hazelcast-worker.yaml文件中。

hazelcast-master.yaml的配置如下所示：

hazelcast: cluster-name: seatunnel-cluster network: rest-api: enabled: true endpoint-groups: CLUSTER_WRITE: enabled: true DATA: enabled: true join: kubernetes: enabled: true service-dns: seatunnel-cluster.bigdata.svc.cluster.local service-port: 5801 port: auto-increment: false port: 5801 properties: hazelcast.invocation.max.retry.count: 20 hazelcast.tcp.join.port.try.count: 30 hazelcast.logging.type: log4j2 hazelcast.operation.generic.thread.count: 50 hazelcast.heartbeat.failuredetector.type: phi-accrual hazelcast.heartbeat.interval.seconds: 30 hazelcast.max.no.heartbeat.seconds: 300 hazelcast.heartbeat.phiaccrual.failuredetector.threshold: 15 hazelcast.heartbeat.phiaccrual.failuredetector.sample.size: 200 hazelcast.heartbeat.phiaccrual.failuredetector.min.std.dev.millis: 200

上述配置文件中的关键配置项如下：

cluster-name

该配置用于确定多个节点是否属于同一个集群，即只有相同cluster-name的节点才会属于同一个集群。如果两个节点之间的cluster-name名称不同，Hazelcast 将会拒绝服务请求。

网络配置
  • rest-api.enabled：在ST 2.3.10版本中Hazelcast REST 服务默认在配置中禁用，需要手动显式指定开启。
  • service-dns（必填）：Headless Service 的完整域名，通常为 ${SERVICE-NAME}.${NAMESPACE}.svc.cluster.local。
  • service-port（可选）：Hazelcast 端口；如果指定的值大于 0，则覆盖默认值（默认端口 = 5701）

使用上述基于k8s的join机制，在Hazelcast Pod启动时会解析service-dns，获取所有成员pod的IP列表（通过Headless Service），然后成员之间通过5801端口尝试建立TCP连接。

同样的，对于hazelcast-worker.yaml配置文件如下所示：

hazelcast: cluster-name: seatunnel-cluster network: rest-api: enabled: true endpoint-groups: CLUSTER_WRITE: enabled: true DATA: enabled: true join: kubernetes: enabled: true service-dns: seatunnel-cluster.bigdata.svc.cluster.local service-port: 5801 port: auto-increment: false port: 5801 properties: hazelcast.invocation.max.retry.count: 20 hazelcast.tcp.join.port.try.count: 30 hazelcast.logging.type: log4j2 hazelcast.operation.generic.thread.count: 50 hazelcast.heartbeat.failuredetector.type: phi-accrual hazelcast.heartbeat.interval.seconds: 30 hazelcast.max.no.heartbeat.seconds: 300 hazelcast.heartbeat.phiaccrual.failuredetector.threshold: 15 hazelcast.heartbeat.phiaccrual.failuredetector.sample.size: 200 hazelcast.heartbeat.phiaccrual.failuredetector.min.std.dev.millis: 200 member-attributes: rule: type: string value: worker

通过上述流程，我们就创建好了与Hazelcast集群相关的配置和服务，实现了Hazecast基于Kubernetes的集群成员发现。

接下来，继续完成有关SeaTunnel引擎的相关配置。

配置SeaTunnel引擎

SeaTunnel引擎的相关配置都在seatunnel.yaml文件中，下面给出seatunnel.yaml配置示例以供参考：

seatunnel: engine: history-job-expire-minutes: 1440 backup-count: 1 queue-type: blockingqueue print-execution-info-interval: 60 print-job-metrics-info-interval: 60 classloader-cache-mode: true http: enable-http: true port: 8080 enable-dynamic-port: false port-range: 100 slot-service: dynamic-slot: true checkpoint: interval: 300000 timeout: 60000 storage: type: hdfs max-retained: 3 plugin-config: namespace: /tmp/seatunnel/checkpoint_snapshot storage.type: hdfs fs.defaultFS: hdfs://xxx:8020 # Ensure that the directory has written permission telemetry: metric: enabled: true

包含以下配置信息：

  • history-job-expire-minutes：任务历史记录保留时长为 24 小时（1440 分钟），超时自动清理。
  • backup-count: 1：任务状态备份副本数为 1。
  • queue-type: blockingqueue：使用阻塞队列管理任务，避免资源耗尽。
  • print-execution-info-interval: 60：每分钟打印一次任务执行状态。
  • print-job-metrics-info-interval: 60：每分钟输出一次任务指标（如吞吐量、延迟）。
  • classloader-cache-mode: true：启用类加载缓存，减少重复加载开销，提升性能。
  • dynamic-slot: true：允许根据负载动态调整任务槽（Slot）数量，优化资源利用率。
  • checkpoint.interval: 300000：每 5 分钟触发一次检查点（Checkpoint）。
  • checkpoint.timeout: 60000：检查点超时时间为 1 分钟。
  • telemetry.metric.enabled: true：启用任务运行指标采集（如延迟、吞吐量），便于监控。

创建k8s yaml文件部署应用

在完成上面的工作流程后，我们就可以进入到最后一步：创建Master和Worker节点的k8s yaml文件定义部署的相关配置。

为了将配置文件与应用程序解耦，我们将上文中列出的配置文件合并到一个ConfigMap中，并挂载到容器的配置路径下，便于对配置文件的统一管理和更新。

以下是针对 seatunnel-cluster-master.yaml 和 seatunnel-cluster-worker.yaml 的配置示例，涵盖了配置 ConfigMap 挂载、容器启动命令以及部署资源定义等相关内容。

seatunnel-cluster-master.yaml：

apiVersion: apps/v1 kind: Deployment metadata: name: seatunnel-cluster-master spec: replicas: 2 # modify replicas according to your case strategy: type: RollingUpdate rollingUpdate: maxUnavailable: 25% maxSurge: 50% selector: matchLabels: app.kubernetes.io/instance: seatunnel-cluster-app app.kubernetes.io/version: 2.3.10 app.kubernetes.io/name: seatunnel-cluster-master app.kubernetes.io/component: master template: metadata: annotations: prometheus.io/path: /hazelcast/rest/instance/metrics prometheus.io/port: "5801" prometheus.io/scrape: "true" prometheus.io/role: "seatunnel-master" labels: app.kubernetes.io/instance: seatunnel-cluster-app app.kubernetes.io/version: 2.3.10 app.kubernetes.io/name: seatunnel-cluster-master app.kubernetes.io/component: master spec: affinity: nodeAffinity: requiredDuringSchedulingIgnoredDuringExecution: nodeSelectorTerms: - matchExpressions: - key: nodeAffinity-key operator: Exists containers: - name: seatunnel-master image: seatunnel:2.3.10 imagePullPolicy: IfNotPresent ports: - containerPort: 5801 name: hazelcast - containerPort: 8080 name: "master-port" command: - /opt/seatunnel/bin/seatunnel-cluster.sh - -r - master resources: requests: cpu: "1" memory: 4G volumeMounts: - mountPath: "/opt/seatunnel/config/hazelcast-master.yaml" name: seatunnel-configs subPath: hazelcast-master.yaml - mountPath: "/opt/seatunnel/config/hazelcast-worker.yaml" name: seatunnel-configs subPath: hazelcast-worker.yaml - mountPath: "/opt/seatunnel/config/seatunnel.yaml" name: seatunnel-configs subPath: seatunnel.yaml - mountPath: "/opt/seatunnel/config/hazelcast-client.yaml" name: seatunnel-configs subPath: hazelcast-client.yaml - mountPath: "/opt/seatunnel/config/log4j2_client.properties" name: seatunnel-configs subPath: log4j2_client.properties - mountPath: "/opt/seatunnel/config/log4j2.properties" name: seatunnel-configs subPath: log4j2.properties volumes: - name: seatunnel-configs configMap: name: seatunnel-cluster-configs
部署策略
  • 采用多副本（replicas=2）部署确保服务高可用
  • 滚动更新策略（RollingUpdate）实现零停机部署：
    • maxUnavailable: 25%：保证更新期间至少75%的Pod保持运行
    • maxSurge: 50%：允许临时增加50%的Pod资源用于平滑过渡
标签选择器
  • 采用Kubernetes推荐的标准标签体系
  • spec.selector.matchLabels： 根据标签定义Deployment管理Pod的范围
  • spec.template.labels： 定义新创建Pod的标签，标识Pod的元数据。
节点亲和性
  • 配置affinity属性指定Pod调度的节点，需要根据自己k8s环境的节点标签进行替换。
配置文件挂载
  • 核心配置文件统一管理在ConfigMap中，便于管理以及与应用程序解耦
  • 通过subPath指定挂载的单个文件

seatunnel-cluster-worker.yaml配置文件如下：

apiVersion: apps/v1 kind: Deployment metadata: name: seatunnel-cluster-worker spec: replicas: 3 # modify replicas according to your case strategy: type: RollingUpdate rollingUpdate: maxUnavailable: 25% maxSurge: 50% selector: matchLabels: app.kubernetes.io/instance: seatunnel-cluster-app app.kubernetes.io/version: 2.3.10 app.kubernetes.io/name: seatunnel-cluster-worker app.kubernetes.io/component: worker template: metadata: annotations: prometheus.io/path: /hazelcast/rest/instance/metrics prometheus.io/port: "5801" prometheus.io/scrape: "true" prometheus.io/role: "seatunnel-worker" labels: app.kubernetes.io/instance: seatunnel-cluster-app app.kubernetes.io/version: 2.3.10 app.kubernetes.io/name: seatunnel-cluster-worker app.kubernetes.io/component: worker spec: affinity: nodeAffinity: requiredDuringSchedulingIgnoredDuringExecution: nodeSelectorTerms: - matchExpressions: - key: nodeAffinity-key operator: Exists containers: - name: seatunnel-worker image: seatunnel:2.3.10 imagePullPolicy: IfNotPresent ports: - containerPort: 5801 name: hazelcast command: - /opt/seatunnel/bin/seatunnel-cluster.sh - -r - worker resources: requests: cpu: "1" memory: 10G volumeMounts: - mountPath: "/opt/seatunnel/config/hazelcast-master.yaml" name: seatunnel-configs subPath: hazelcast-master.yaml - mountPath: "/opt/seatunnel/config/hazelcast-worker.yaml" name: seatunnel-configs subPath: hazelcast-worker.yaml - mountPath: "/opt/seatunnel/config/seatunnel.yaml" name: seatunnel-configs subPath: seatunnel.yaml - mountPath: "/opt/seatunnel/config/hazelcast-client.yaml" name: seatunnel-configs subPath: hazelcast-client.yaml - mountPath: "/opt/seatunnel/config/log4j2_client.properties" name: seatunnel-configs subPath: log4j2_client.properties - mountPath: "/opt/seatunnel/config/log4j2.properties" name: seatunnel-configs subPath: log4j2.properties volumes: - name: seatunnel-configs configMap: name: seatunnel-cluster-configs

定义好上述master和worker的yaml文件后，就可以执行以下命令进行部署到k8s集群了：

kubectl apply -f seatunnel-cluster-master.yaml kubectl apply -f seatunnel-cluster-worker.yaml

正常情况下会看到SeaTunnel集群中共有2个master节点和3个worker节点：

$ kubectl get pods | grep seatunnel-cluster seatunnel-cluster-master-6989898f66-6fjz8 1/1 Running 0 156m seatunnel-cluster-master-6989898f66-hbtdn 1/1 Running 0 155m seatunnel-cluster-worker-87fb469f7-5c96x 1/1 Running 0 156m seatunnel-cluster-worker-87fb469f7-7kt2h 1/1 Running 0 155m seatunnel-cluster-worker-87fb469f7-drm9r 1/1 Running 0 156m

至此，我们已成功在Kubernetes环境中以分离集群模式部署了SeaTunnel集群。

如今，集群已就绪，如何在客户端向其提交任务呢？

客户端提交任务到集群

使用命令行工具提交任务

有关SeaTunnel客户端的配置都在hazelcast-client.yaml文件中。

首先需要在客户端本地下载二进制安装包（包含bin、config文件），并保证SeaTunnel的安装路径与服务端一致，这也就是官网中所说的：Setting the&nbsp;SEATUNNEL_HOME&nbsp;the same as the server，否则，可能会导致出现诸如无法在服务器端找到连接器插件路径等错误（因为服务端插件路径与客户端路径不一致）。

进入安装路径下，只需要修改config/hazelcast-client.yaml文件，配置指向刚刚创建的Headless Service服务地址即可：

hazelcast-client: cluster-name: seatunnel-cluster properties: hazelcast.logging.type: log4j2 connection-strategy: connection-retry: cluster-connect-timeout-millis: 3000 network: cluster-members: - seatunnel-cluster.bigdata.svc.cluster.local:5801

客户端配置完成后，即可将任务提交至集群执行。任务提交时的JVM参数配置方式主要有两种：

  • 在config/jvm_client_options文件中配置任务提交时的JVM参数

    此方法配置的JVM参数将应用于所有通过seatunnel.sh提交的任务，无论运行于本地模式还是集群模式。所有提交的任务都将共享相同的JVM参数配置。

  • 在提交任务的命令行中指定JVM参数。

    使用seatunnel.sh提交任务时，可在命令行中直接指定JVM参数，例如：sh bin/seatunnel.sh --config $SEATUNNEL_HOME/config/v2.batch.config.template -DJvmOption=-Xms2G -Xmx2G。此方法允许为每个提交的任务独立配置JVM参数。

接下来通过一个案例来演示客户端提交任务至集群执行的完整流程：

env { parallelism = 2 job.mode = "STREAMING" checkpoint.interval = 2000 } source { FakeSource { parallelism = 2 plugin_output = "fake" row.num = 16 schema = { fields { name = "string" age = "int" } } } } sink { Console { } }

在客户端使用以下命令提交任务：

sh bin/seatunnel.sh --config config/v2.streaming.example.template -m cluster -n st.example.template -DJvmOption="-Xms2G -Xmx2G"

在Master节点，使用如下命令列出正在运行的任务列表：

$ sh bin/seatunnel.sh -l Job ID Job Name Job Status Submit Time Finished Time ------------------ ------------------- ---------- ----------------------- ----------------------- 964354250769432580 st.example.template RUNNING 2025-04-15 10:39:30.588

可以看到，我们刚刚向集群中提交的st.example.template任务已经处于RUNNING状态了。现在我们可以在Worker节点日志中看到如下日志打印：

2025-04-15 10:34:41,998 INFO [.a.s.c.s.c.s.ConsoleSinkWriter] [st-multi-table-sink-writer-1] - subtaskIndex=0 rowIndex=1: SeaTunnelRow#tableId=fake SeaTunnelRow#kind=INSERT : bdaUB, 110348049 2025-04-15 10:34:41,998 INFO [.a.s.c.s.c.s.ConsoleSinkWriter] [st-multi-table-sink-writer-1] - subtaskIndex=1 rowIndex=1: SeaTunnelRow#tableId=fake SeaTunnelRow#kind=INSERT : mOifY, 1974539087 2025-04-15 10:34:41,999 INFO [.a.s.c.s.c.s.ConsoleSinkWriter] [st-multi-table-sink-writer-1] - subtaskIndex=0 rowIndex=2: SeaTunnelRow#tableId=fake SeaTunnelRow#kind=INSERT : jKFrR, 1828047742 2025-04-15 10:34:41,999 INFO [.a.s.c.s.c.s.ConsoleSinkWriter] [st-multi-table-sink-writer-1] - subtaskIndex=1 rowIndex=2: SeaTunnelRow#tableId=fake SeaTunnelRow#kind=INSERT : gDiqR, 1177544796 2025-04-15 10:34:41,999 INFO [.a.s.c.s.c.s.ConsoleSinkWriter] [st-multi-table-sink-writer-1] - subtaskIndex=0 rowIndex=3: SeaTunnelRow#tableId=fake SeaTunnelRow#kind=INSERT : bCVxc, 909343602 ...

说明我们的任务成功提交至所创建的SeaTunnel集群，并且确认其正常运行。

使用Rest Api接口提交任务

SeaTunnel提供了通过Rest Api接口的方式来查询运行作业的状态和统计信息，以及提交/停止作业等操作。

在上文中我们配置了只包含Master节点的Headless Service，并指定暴露的端口为8080。因此，我们就可以在客户端使用Rest API接口的方式来实现任务的提交。

SeaTunnel Rest API接口提供了通过上传配置文件来提交任务，命令如下：

 $ curl 'http://seatunnel-cluster-master.bigdata.svc.cluster.local:8080/submit-job/upload' --form 'config_file=@"/opt/seatunnel/config/v2.streaming.example.template"' --form 'jobName=st.example.template' {"jobId":"964553575034257409","jobName":"st.example.template"}

如果作业提交成功，会返回jobId和jobName，如上所示。

接下来，通过Rest API接口获取集群正在运行的所有任务，观察刚刚提交的任务信息：

curl 'http://seatunnel-cluster-master.bigdata.svc.colo.gzgalocal:8080/running-jobs' [{"jobId":"964553575034257409","jobName":"st.example.template","jobStatus":"RUNNING","envOptions":{"job.mode":"STREAMING","checkpoint.interval":"2000","parallelism":"2"}, ...]

可以看到接口返回显示了任务状态和其他额外的元数据信息，说明我们通过Rest Api接口提交任务的方式也成功执行。更多Rest Api接口介绍可以参考官网：RESTful API V2

总结

本文着重介绍了如何以推荐的分离集群模式（Separated Cluster Mode）部署k8s集群的实践，总结下来，部署过程主要包含以下步骤：

  1. 准备 Kubernetes 环境

    确保已搭建并运行一个可用的 Kubernetes 集群，并安装所有必要的组件。

  2. 构建 SeaTunnel Docker 镜像

    如果没有二次开发需求，可直接使用官方提供的镜像。否则，在本地编译打包后，编写 Dockerfile 并构建 SeaTunnel 镜像。

  3. 配置Headless Service和Hazelcast集群

    Hazelcast的k8s自动发现机制的DNS Lookup模式是基于k8s的Headless Service功能来实现的，因此首先创建Headless Service服务，并在hazelcast的yaml配置文件中通过service-dns来指定服务地址。

    Headless Service会在域名解析时解析成所包含pod的IP地址集合，以此实现hazelcast集群成员之间的彼此发现。

  4. 配置 SeaTunnel 引擎

    修改seatunnel.yaml文件，配置SeaTunnel引擎参数。

  5. 创建k8s yaml部署文件

    分别创建Master和Worker的k8s yaml文件，配置节点标签、启动命令、资源和数据卷挂载等内容，最终将其部署到k8s集群。

  6. 配置 SeaTunnel 客户端

    在客户端安装SeaTunnel，并确保客户端的安装路径 (SEATUNNEL_HOME) 与服务端一致。修改 hazelcast-client.yaml 文件，配置客户端连接到集群Service服务的地址。

  7. 任务提交与执行:

    完成以上步骤后，即可在客户端提交任务并由 SeaTunnel 集群执行。

本文上述配置案例仅供参考，可能仍有很多配置项和配置内容未涉及，欢迎各位补充与讨论，希望有各位有所帮助！ > 本文由 白鲸开源科技 提供发布支持！</version_tag>

原文链接：https://my.oschina.net/SeaTunnel/blog/18219094
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有，本站原创内容转载请注明来源。

相关文章

    文章评论

    共有0条评论来说两句吧...

    文章二维码

    扫描即可查看该文章

    点击排行

    推荐阅读

    最新文章