Apache SeaTunnel 2.3.10 源码解析：Zeta 引擎服务启动-低调大师

Apache SeaTunnel 2.3.10 源码解析：Zeta 引擎服务启动

2025-12-26 27

今天主要看SeaTunnel自研的数据同步引擎，叫Zeta。首先，如果使用的是zeta引擎，那么第一步一定是运行bin/seatunnel-cluster.sh脚本，这个脚本就是启动zeta的服务端的。

打开seatunnel-cluster.sh看看，可以看到其实是去启动seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/SeaTunnelServer.java中的main()方法

这个就是zeta的核心启动方法了。

如下方代码所示：

public class SeaTunnelServer {

    public static void main(String[] args) throws CommandException {

        ServerCommandArgs serverCommandArgs =
                CommandLineUtils.parse(
                        args,
                        new ServerCommandArgs(),
                        EngineType.SEATUNNEL.getStarterShellName(),
                        acceptUnknownOptions: true
                );

        SeaTunnel.run(serverCommandArgs.buildCommand());
    }
}

其实应该先看看ServerCommandArgs类，该类会基于命令参数拼装启动类，方法入口为serverCommandArgs.buildCommand()

@EqualsAndHashCode(callSuper = true)
@Data
public class ServerCommandArgs extends CommandArgs {

    @Parameter(
            names = {"-cn", "--cluster"},
            description = "The name of cluster"
    )
    private String clusterName;

    @Parameter(
            names = {"-d", "--daemon"},
            description = "The cluster daemon mode"
    )
    private boolean daemonMode = false;

    @Parameter(
            names = {"-r", "--role"},
            description =
                    "The cluster node role, default is master_and_worker, " +
                    "support master, worker, master_and_worker"
    )
    private String clusterRole;

    @Override
    public Command<?> buildCommand() {
        return new ServerExecuteCommand(this);
    }
}

接着SeaTunnel.run()会启动SeaTunnelServer，启动流程如下：

共分为以下步骤：

1）校验当前环境 2）加载SeaTunnel配置 3）设置节点角色，包含Master、Worker、Master_And_Worker 4）创建Hazelcast实例，用于集群发现、注册、分布式数据管理等

@Override
public void execute() {
    // Validate environment
    checkEnvironment();

    // Load SeaTunnel configuration
    SeaTunnelConfig seaTunnelConfig = ConfigProvider.locateAndGetSeaTunnelConfig();
    String clusterRole = this.serverCommandArgs.getClusterRole();

    // Set node role
    if (StringUtils.isNotBlank(clusterRole)) {
        if (EngineConfig.ClusterRole.MASTER.toString().equalsIgnoreCase(clusterRole)) {
            seaTunnelConfig.getEngineConfig().setClusterRole(EngineConfig.ClusterRole.MASTER);
        } else if (EngineConfig.ClusterRole.WORKER.toString().equalsIgnoreCase(clusterRole)) {
            seaTunnelConfig.getEngineConfig().setClusterRole(EngineConfig.ClusterRole.WORKER);
            // In Hazelcast lite node, it will not store IMap data.
            seaTunnelConfig.getHazelcastConfig().setLiteMember(true);
        } else {
            throw new SeaTunnelEngineException("Not supported cluster role: " + clusterRole);
        }
    } else {
        seaTunnelConfig.getEngineConfig().setClusterRole(EngineConfig.ClusterRole.MASTER_AND_WORKER);
    }

    // Create Hazelcast instance for cluster discovery, registration, and distributed data management
    SeaTunnelServerStarter.createHazelcastInstance(seaTunnelConfig, Thread.currentThread().getName());
}

节点角色说明：

Master 节点：

核心职责：负责集群的作业调度、状态管理及资源协调。Master 节点运行 CoordinatorService 服务，处理作业的逻辑计划（LogicalDAG）到物理计划（PhysicalDAG）的转换，并生成执行计划。此外，它还管理检查点（Checkpoint）机制和作业监控指标。
高可用性：采用 Active/Standby 模式，同一时间仅有一个 Active Master，其余为 Standby。当 Active Master 故障时，会触发选举新 Master，确保集群持续运行。
数据存储：通过内置的分布式内存网格（如 Hazelcast IMap）存储作业状态和元数据，无需依赖外部系统（如 ZooKeeper）。在分离部署模式下，所有状态数据仅存储在 Master 节点，避免 Worker 节点负载影响数据稳定性。

Worker 节点：

核心职责：执行具体的数据处理任务。Worker 节点运行 TaskExecutionService 和 SlotService，前者提供任务运行时环境，后者管理节点的资源分配（如 CPU 核心数）。
动态资源分配：通过 SlotService 实现资源的动态划分，支持按任务并行度动态调整资源，提升资源利用率。
无状态设计：Worker 节点不存储作业状态数据，仅负责计算。在容错场景下，Worker 故障后任务会被重新调度到其他节点，依赖 Master 存储的状态恢复。

混合角色节点（旧架构）：

在早期版本中，节点角色未严格分离，同一节点可同时作为 Master 和 Worker（称为 master_and_worker 模式）。此架构下，节点既参与调度又执行任务，但在高负载场景下可能导致容错效率问题（如主节点故障引发连锁负载压力）。
新架构优化：2.3.6 版本后推荐分离部署模式，彻底解耦 Master 与 Worker 角色，提升集群稳定性和扩展性。

接着回到上面的创建Hazelcast实例：

如下所示，核心代码为：HazelcastInstanceFactory.newHazelcastInstance方法，该方法表示创建了Hazelcast实例；

private static HazelcastInstanceImpl initializeHazelcastInstance(
        @NonNull SeaTunnelConfig seaTunnelConfig, String customInstanceName) {

    // Set the default async executor for Hazelcast InvocationFuture
    ConcurrencyUtil.setDefaultAsyncExecutor(CompletableFuture.EXECUTOR);

    // Check whether to enable metrics reporting
    boolean condition = checkTelemetryConfig(seaTunnelConfig);

    String instanceName = customInstanceName != null
            ? customInstanceName
            : HazelcastInstanceFactory.createInstanceName(seaTunnelConfig.getHazelcastConfig());

    // Create Hazelcast instance
    HazelcastInstanceImpl original = ((HazelcastInstanceProxy)
            HazelcastInstanceFactory.newHazelcastInstance(
                    seaTunnelConfig.getHazelcastConfig(),
                    instanceName,
                    new SeaTunnelNodeContext(seaTunnelConfig)))
            .getOriginal();

    // Initialize telemetry instance
    // Enable metrics reporting, including: JvmCollector, JobInfoDetail, ThreadPoolStatus, NodeMetrics, ClusterMetrics
    if (condition) {
        initTelemetryInstance(original.node);
    }

    return original;
}

其中，最重要的就是这句代码中的new SeaTunnelNodeContext(seaTunnelConfig)，这里会返回一个SeaTunnelNodeContext类，这个类是继承自Hazelcast这个组件的DefaultNodeContext类。在Hazelcast启动的过程中，会去调用DefaultNodeContext类的实现类的createNodeExtension（）方法，在这里其实也就是SeaTunnelNodeContext类的createNodeExtension（）方法。这里不具体展开讲解Hazelcast类，大家可以去查一下其他Hazelcast资料。

然后我们接着分析在Hazelcast节点启动时会调用createNodeExtension方法

@Slf4j
public class SeaTunnelNodeContext extends DefaultNodeContext {

    private final SeaTunnelConfig seaTunnelConfig;

    public SeaTunnelNodeContext(@NonNull SeaTunnelConfig seaTunnelConfig) {
        this.seaTunnelConfig = seaTunnelConfig;
    }

    @Override
    public NodeExtension createNodeExtension(@NonNull Node node) {
        return new org.apache.seatunnel.engine.server.NodeExtension(node, seaTunnelConfig);
    }
}

这里跟踪进去，查看节点扩展的实现，这里初始化了Zeta引擎。

SeaTunnelServe类实现了一系列Hazelcast接口，用于监听集变更状态，包含：节点初始化、集群节点加入/移除，跟踪和管理分布式系统操作；

接下来依次分析各个操作：

1）节点初始化：

这个初始化方法为SeaTunnel服务器提供了完整的启动流程，确保了各个服务组件的正确初始化和配置；

核心方法包括： startMaster()方法和startWorker()方法

@Override
public void init(NodeEngine engine, Properties hzProperties) {
    this.nodeEngine = (NodeEngineImpl) engine;

    // TODO: Decide whether to run these methods on the master node based on deployment type
    classLoaderService = new DefaultClassLoaderService(
            seaTunnelConfig.getEngineConfig().isClassLoaderCacheMode(), nodeEngine);

    // Handles event processing and forwarding
    eventService = new EventService(nodeEngine);

    // Start Master / Worker capabilities
    if (EngineConfig.ClusterRole.MASTER_AND_WORKER.ordinal()
            == seaTunnelConfig.getEngineConfig().getClusterRole().ordinal()) {
        startWorker();
        startMaster();
    } else if (EngineConfig.ClusterRole.WORKER.ordinal()
            == seaTunnelConfig.getEngineConfig().getClusterRole().ordinal()) {
        startWorker();
    } else {
        startMaster();
    }

    // SeaTunnel health-check monitor
    seaTunnelHealthMonitor = new SeaTunnelHealthMonitor(((NodeEngineImpl) engine).getNode());

    // Task log management service
    if (seaTunnelConfig.getEngineConfig().getTelemetryConfig() != null
            && seaTunnelConfig.getEngineConfig().getTelemetryConfig().getLogs() != null
            && seaTunnelConfig.getEngineConfig().getTelemetryConfig().getLogs().isEnabled()) {
        taskLogManagerService = new TaskLogManagerService(
                seaTunnelConfig.getEngineConfig().getTelemetryConfig().getLogs());
        taskLogManagerService.initClean();
    }

    // Start Jetty service: provides HTTP REST API, Web UI, job management, monitoring endpoints, etc.
    if (seaTunnelConfig.getEngineConfig().getHttpConfig().isEnabled()) {
        jettyService = new JettyService(nodeEngine, seaTunnelConfig);
        jettyService.createJettyServer();
    }

    // A trick to fix StatisticsDataReference cleaner thread class-loader leak.
    // See https://issues.apache.org/jira/browse/HADOOP-19049
    FileSystem.Statistics statistics = new FileSystem.Statistics("SeaTunnel");
}

首先是startMaster方法，主要初始化了协调器服务、检查点服务、监控服务；

各个服务职责如下：

协调器服务：负责作业调度管理、集群资源协同、任务分配、状态同步等；
检查点服务：检查点管理、数据一致性保证、故障恢复支持、状态保存和恢复等；

监控服务：定期打印执行信息、监控系统状态、性能指标收集；

private void startMaster() { // Initialize coordinator service coordinatorService = new CoordinatorService(nodeEngine, this, seaTunnelConfig.getEngineConfig());

  // Initialize checkpoint service
  checkpointService = new CheckpointService(seaTunnelConfig.getEngineConfig().getCheckpointConfig());

  // Initialize monitoring service
  monitorService = Executors.newSingleThreadScheduledExecutor();
  monitorService.scheduleAtFixedRate(
          this::printExecutionInfo,
          0,                                      // initial delay
          seaTunnelConfig.getEngineConfig().getPrintExecutionInfoInterval(),
          TimeUnit.SECONDS);

}

其次是startWorker方法，这里重点介绍TaskExecutionService服务，通过这个服务，SeaTunnel能够高效地执行各种数据处理任务，同时保证系统的稳定性和可靠性；

private void startWorker() {
    // 1. Initialize task execution service
    taskExecutionService = new TaskExecutionService(classLoaderService, nodeEngine, eventService);

    // 2. Register metrics collector
    nodeEngine.getMetricsRegistry().registerDynamicMetricsProvider(taskExecutionService);

    // 3. Start task execution service
    taskExecutionService.start();

    // 4. Initialize slot service
    getSlotService();
}

其实在SeaTunnelServer最核心的就是调用TaskExecutionService类的start()方法，基本流程如下图所示：

这里引用来自一篇官方文章的介绍文字：

TaskExecutionService TaskExecutionService 是一个执行任务的服务，将在每个节点上运行一个实例。它从 JobMaster 接收 TaskGroup 并在其中运行 Task。并维护TaskID->TaskContext，对Task的具体操作都封装在TaskContext中。而Task内部持有OperationService，也就是说Task可以通过OperationService远程调用其他Task或JobMaster进行通信。

CoordinatorService CoordinatorService是一个充当协调器的服务，它主要负责处理客户端提交的命令以及切换master后任务的恢复。客户端在提交任务时会找到master节点并将任务提交到CoordinatorService服务上，CoordinatorService会缓存任务信息并等待任务执行结束。当任务结束后再对任务进行归档处理。

SlotService SlotService是slot管理服务，用于管理集群的可用Slot资源。SlotService运行在所有节点上并定期向master上报资源信息。

2）集群节点加入/移除：

memberAdded：处理集群成员加入事件

memberRemoved：处理集群成员离开事件

这里说明下问什么memberAdded是空实现：

a) 设计考虑：

成员加入是一个正常的事件，不需要特殊处理；

新成员加入时会自动进行初始化；

资源分配和任务调度是动态的；

b）实际原因：

新成员加入时，会通过其他机制（如SlotService）自动处理资源分配；

任务调度是动态的，不需要在成员加入时特别处理；

保持简单性，避免不必要的复杂性；

memberRemoved主要处理逻辑：只在主节点上处理成员移除事件，成员移除需要处理的关键问题：

a）资源回收：释放离开节点的资源、重新分配任务、清理相关状态；

b）任务重分配：重新分配离开节点的任务、确保任务继续执行、维护任务状态

c）状态维护：更新集群状态、维护成员列表、更新资源分配；

为什么需要memberRemoved：

a）可靠性考虑：节点离开可能影响任务执行、需要确保数据一致性、需要保证系统可用性；

b）资源管理：需要及时释放资源、需要重新分配任务、需要维护集群状态；

3）跟踪和管理分布式系统操作：

实现是空的，说明在当前版本没有特别需要跟踪的操作；

@Override
public void populate(LiveOperations liveOperations) {
    // In SeaTunnelServer this implementation is empty,
    // indicating the current version has no special operations to track.
}

最后，分享一个在本地Idea环境启动过程中的问题：

如下所示，官方默认配置为hdfs方式，由于本地缺少hdfs环境，因此会阻碍服务启动，调整为localfile本地即可启动。

最后，访问localhost:8080即可查看服务状态：

微信关注我们

原文链接：https://my.oschina.net/SeaTunnel/blog/19107089

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

PAG在得物社区S级活动的落地

一、背景近期，得物社区活动「用篮球认识我」推出 "用户上传图片生成专属球星卡" 核心玩法。初期规划由服务端基于 PAG 技术合成，为了让用户可以更自由的定制专属球星卡，经多端评估后确定：由 H5 端承接 "图片交互调整 - 球星卡生成" 核心链路，支持用户单指拖拽、双指缩放 / 旋转人像，待调整至理想位置后触发合成。而 PAG 作为腾讯自研开源的动效工作流解决方案，凭借跨平台渲染一致性、图层实时编辑、轻量化文件性能，能精准匹配需求，成为本次核心技术选型。鉴于 H5 端需落地该核心链路，且流程涉及 PAG 技术应用，首先需对 PAG 技术进行深入了解，为后续开发与适配奠定基础。二、PAG是什么？这里简单介绍一下，PAG 是腾讯自研并开源的动效工作流解决方案，核心是实现 Adobe After Effects（AE）动效的一键导出与跨平台应用，包含渲染 SDK、AE 导出插件（PAGExporter）、桌面预览工具（PAGViewer）三部分。它导出的二进制 PAG 文件压缩率高、解码快，能集成多类资源；支持 Android、iOS、Web 等全平台，且各端渲染一致、开启 GP...

2025-12-26

24

在数字化转型浪潮席卷全球的今天，智能制造已成为制造业高质量发展的核心引擎。然而，在迈向智能化的道路上，企业面临着诸多挑战：多系统数据孤岛、复杂的调度依赖、监控告警滞后等问题层出不穷。在近期的 Apache DolphinScheduler 线上用户交流会上，社区邀请到了深圳某大型智能制造企业高级软件工程师邱忠标，现场为大家带来一场关于 Apache DolphinScheduler 在该企业智能制造业务场景中的应用实践分享。现将活动演讲核心内容整理成为，深入探讨这家企业是如何通过 Apache DolphinScheduler 实现企业调度平台质的飞跃。关于作者邱忠标，深圳某大型智能制造企业高级软件工程师，专注于智能制造领域的数据技术研究与实践，致力于推动制造业数字化转型。智能制造的时代背景随着工业4.0的深入推进，智能制造已成为各国制造业竞争的焦点。智能制造成熟度模型从低到高分为多个层级，企业需要逐步提升自动化、数字化、网络化水平，最终实现智能化生产。在这一进程中，数据成为核心生产要素，如何高效、稳定地采集、处理和调度这些数据，成为摆在每个制造企业面前的难题。现代制造...

2025-12-26

20

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。