SeaTunnel Zeta 引擎 RESTful API V1 完整指南：作业提交、监控查询与运维接口详解

SeaTunnel Zeta 引擎 RESTful API V1 完整指南：作业提交、监控查询与运维接口详解

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel

本篇技术指南基于 SeaTunnel 官方文档，系统讲解 Zeta 引擎内置的 RESTful 监控与运维 API（V1 版本）。读完后你可以掌握：如何在hazelcast.yaml中启用并配置该 API，如何使用全部 REST 接口完成作业提交（单个/批量）、停止、加密配置、集群概览查询、作业详情与指标拉取、Worker 资源快照、线程堆栈导出以及跨节点日志获取，并能结合源码理解各接口在引擎中的实现位置与调用关系，为搭建作业监控看板或自动化运维工具打下基础。

版本提示：V1 版本 API 已弃用，官方推荐使用 REST API V2，且 V1 API 默认已关闭，需要手动在hazelcast.yaml中显式启用后才能使用。

概述与启用方式

SeaTunnel 内置了一套用于监控的 API，可用于查询运行中作业的状态和统计信息、最近完成的作业，同时也支持通过 HTTP 提交与停止作业。监控 API 是 RESTful 风格，接受 HTTP 请求并使用 JSON 数据格式响应。

监控 API 由运行的 Web 服务提供，它是节点进程的一部分：每个节点成员都可以提供 REST API 功能，而不是只有 Master 节点。默认情况下，服务器禁用了 RESTful API V1。

在 hazelcast.yaml 中启用

在 config/hazelcast.yaml 中，rest-api.enabled默认为false，同时endpoint-groups中的CLUSTER_WRITE与DATA端点组默认开启。要启用 V1 API，需要设置rest-api.enabled: true：

network: rest-api: enabled: true endpoint-groups: CLUSTER_WRITE: enabled: true DATA: enabled: true join: tcp-ip: enabled: true member-list: - localhost port: auto-increment: true port-count: 100 port: 5801

服务监听端口为 5801，可在port配置段中修改，并支持auto-increment与port-count用于多节点端口自增分配。所有 V1 接口的统一上下文路径为/hazelcast/rest/maps，这一点可以从源码 RestConstant.java 中得到印证：

public static final String CONTEXT_PATH = "/hazelcast/rest/maps";

认证说明

引擎侧还实现了 Basic 认证过滤器 BasicAuthFilter.java。当http配置开启enableBasicAuth时，所有请求需携带Authorization: Basic <Base64(用户名:密码)>头，校验失败会返回 401 与WWW-Authenticate: Basic realm="SeaTunnel Web UI"响应头；未开启时请求直接放行。

API 接口总览

V1 版本提供以下接口（均位于/hazelcast/rest/maps路径下）：

以上端点常量均定义在 RestConstant.java 中，且每个端点在 servlet 目录 下有对应的 Servlet 实现、在 service 目录 下有对应的 Service 逻辑，可据此进一步阅读源码。

获取 Connector 的 OptionRule

接口：GET /hazelcast/rest/maps/option-rules?type=source&plugin=FakeSource

返回指定 Connector 运行时完整的 OptionRule 元数据，由 OptionRulesService.java 处理，响应结构定义在 OptionRuleResponse.java。

参数

响应

{ "engineType": "seatunnel", "pluginType": "source", "pluginName": "FakeSource", "optionRule": { "optionalOptions": [ { "key": "row.num", "type": "java.lang.Integer", "defaultValue": 5, "description": "The total number of data generated per degree of parallelism", "fallbackKeys": [], "optionValues": null } ], "requiredOptions": [ { "ruleType": "EXCLUSIVE", "options": [ { "key": "schema", "type": "org.apache.seatunnel.api.table.catalog.TableSchema", "defaultValue": null, "description": "The schema of the upstream table", "fallbackKeys": [], "optionValues": null } ] }, { "ruleType": "CONDITIONAL", "options": [ { "key": "string.template", "type": "java.util.List<java.lang.String>", "defaultValue": null, "description": "The template list of string type that connector generated, if user configured it, connector will randomly select an item from the template list", "fallbackKeys": [], "optionValues": null } ], "expression": "'string.fake.mode' == TEMPLATE", "expressionTree": { "condition": { "option": { "key": "string.fake.mode", "type": "org.apache.seatunnel.connectors.seatunnel.fake.config.FakeSourceOptions$FakeMode", "defaultValue": "RANDOM", "description": "The fake mode of generating string data", "fallbackKeys": [], "optionValues": [ "RANDOM", "TEMPLATE" ] }, "expectValue": "TEMPLATE", "operator": null, "next": null }, "operator": null, "next": null } } ], "conditionRules": [] } }

说明：

• 响应结果来自运行时 plugin discovery，会跟随服务端实际安装的 connector 版本。
• requiredOptions[].ruleType可能是ABSOLUTELY_REQUIRED、EXCLUSIVE、BUNDLED或CONDITIONAL。
• optionRule.conditionRules会递归返回嵌套条件规则；当 connector 未定义嵌套规则时，该字段返回空数组。
• 对于条件规则，会同时返回expression和expressionTree，便于 Web 端做动态表单渲染——这是构建可视化作业配置工具（如 Web UI 的表单自动补全）的关键接口。

返回 Zeta 集群的概览

接口：GET /hazelcast/rest/maps/overview?tag1=value1&tag2=value2

参数

响应

{ "projectVersion":"2.3.10-SNAPSHOT", "gitCommitAbbrev":"DeadD0d0", "totalSlot":"0", "unassignedSlot":"0", "works":"1", "runningJobs":"0", "finishedJobs":"0", "failedJobs":"0", "cancelledJobs":"0" }

注意：

• 当你使用dynamic-slot时，返回结果中的totalSlot和unassignedSlot将始终为 0。设置为固定的 slot 值后，将正确返回集群中总共的 slot 数量以及未分配的 slot 数量。
• 当添加标签过滤后，works、totalSlot、unassignedSlot将返回满足条件的节点的相关指标。注意runningJobs等 job 相关指标为集群级别结果，无法根据标签进行过滤。

源码实现

从源码结构看，OverviewService.java 的处理逻辑为：若当前节点即为 Master，则直接本地取概览信息；否则通过NodeEngineUtil.sendOperationToMasterNode将GetOverviewOperation发送到了 Master 节点执行，再补充projectVersion与gitCommitAbbrev版本信息后返回。值得注意的是，当 Master 选举尚未完成（如本地引擎刚启动）时，服务会返回空的OverviewInfo而非报错，客户端此时轮询即可。

查询 Worker 资源

接口：GET /hazelcast/rest/maps/resource/workers

返回已注册 Worker 的当前资源快照，无请求参数。

响应

{ "available": true, "collectedAt": 1723017600000, "workers": [ { "address": "10.0.0.8:5801", "tags": {"region": "us-west"}, "totalSlots": 4, "freeSlots": 1, "usedSlots": 3, "dynamicSlot": false, "totalCpuCores": 8, "availableCpuCores": 2, "totalHeapMemoryBytes": 17179869184, "availableHeapMemoryBytes": 4294967296, "cpuUsage": 0.42, "memUsage": 0.58, "runningJobIds": [123456789] } ] }

说明：

• 固定 Slot 模式的 Worker 返回totalSlots、usedSlots和freeSlots。
• 动态 Slot 模式的 Worker 没有固定的 Slot 容量。此时，totalSlots表示当前已跟踪的已分配和未分配 Slot 总数，freeSlots表示当前未分配数量。解释容量时，请结合dynamicSlot以及 CPU 和堆内存字段。
• 当available为false时，表示当前无法读取 Master 资源快照，workers为空。客户端应重试，而不应将该响应解释为空集群。
• collectedAt表示 Master 构建本次响应的时间。Worker 字段来自资源管理器收到的最近一次心跳，并不与/system-monitoring-information构成原子快照。
• 如果最近一次 Worker 心跳尚未包含资源或使用率数据，对应字段不会返回。

返回当前节点的线程堆栈信息

接口：GET /hazelcast/rest/maps/thread-dump

无请求参数，返回当前节点（即接收请求的那个节点）的线程堆栈信息，适合在作业出现卡死、死锁等异常时远程诊断，等价于对节点执行一次jstack。

响应

[ { "threadName": "", "threadId": 0, "threadState": "", "stackTrace": "" } ]

返回所有作业及其当前状态的概览

接口：GET /hazelcast/rest/maps/running-jobs

无请求参数。由 RunningJobsServlet.java 与 JobInfoService.java 处理，返回所有作业及其当前状态的概览列表。

响应

[ { "jobId": "", "jobName": "", "jobStatus": "", "envOptions": { }, "createTime": "", "jobDag": { "jobId": "", "envOptions": [], "vertexInfoMap": [ { "vertexId": 1, "type": "", "vertexName": "", "tablePaths": [ "" ] } ], "pipelineEdges": {} }, "pluginJarsUrls": [ ], "isStartWithSavePoint": false, "metrics": { "sourceReceivedCount": "", "sinkWriteCount": "" } } ]

其中jobDag中的vertexInfoMap描述了执行图的顶点信息（顶点 ID、类型、名称、关联表路径），pipelineEdges描述了 Pipeline 之间的边关系，可用于前端绘制作业 DAG 图。

返回作业的详细信息

接口：GET /hazelcast/rest/maps/job-info/:jobId

参数

响应

{ "jobId": "", "jobName": "", "jobStatus": "", "createTime": "", "jobDag": { "jobId": "", "envOptions": [], "vertexInfoMap": [ { "vertexId": 1, "type": "", "vertexName": "", "tablePaths": [ "" ] } ], "pipelineEdges": {} }, "metrics": { "SourceReceivedCount": "", "SourceReceivedQPS": "", "SourceReceivedBytes": "", "SourceReceivedBytesPerSeconds": "", "SinkWriteCount": "", "SinkWriteQPS": "", "SinkWriteBytes": "", "SinkWriteBytesPerSeconds": "", "SinkCommittedCount": "", "SinkCommittedQPS": "", "SinkCommittedBytes": "", "SinkCommittedBytesPerSeconds": "", "TableSourceReceivedCount": {}, "TableSourceReceivedBytes": {}, "TableSourceReceivedBytesPerSeconds": {}, "TableSourceReceivedQPS": {}, "TableSinkWriteCount": {}, "TableSinkWriteQPS": {}, "TableSinkWriteBytes": {}, "TableSinkWriteBytesPerSeconds": {}, "TableSinkCommittedCount": {}, "TableSinkCommittedQPS": {}, "TableSinkCommittedBytes": {}, "TableSinkCommittedBytesPerSeconds": {} }, "finishedTime": "", "errorMsg": null, "envOptions": { }, "pluginJarsUrls": [ ], "isStartWithSavePoint": false }

字段返回规则

• jobId、jobName、jobStatus、createTime、jobDag、metrics字段总会返回。
• envOptions、pluginJarsUrls、isStartWithSavePoint字段在 Job 处于 RUNNING 状态时会返回。
• finishedTime、errorMsg字段在 Job 结束时会返回，结束状态为不为 RUNNING，可能为 FINISHED，可能为 CANCEL。
• 运行中的 Job 还会返回diagnostics字段（状态时间戳与各 Pipeline 的恢复次数），该字段只在本接口返回，/running-jobs不返回，更详细的说明可参考 REST API V2。

指标字段说明

按表汇总的指标名常量与RestConstant.java中的TABLE_SOURCE_RECEIVED_COUNT、TABLE_SINK_WRITE_COUNT、TABLE_SINK_COMMITTED_COUNT等定义一一对应。

当我们查询不到这个 Job 时，返回结果为：

{ "jobId" : "" }

已弃用的旧接口：/running-job/:jobId

接口：GET /hazelcast/rest/maps/running-job/:jobId

此 API 已经弃用（源码中该端点常量也标注了@Deprecated），请使用/hazelcast/rest/maps/job-info/:jobId替代。其参数与响应结构同上：jobId（必传，long 类型，job id），响应字段中metrics仅包含sourceReceivedCount、sinkWriteCount等基础指标，同样遵循“RUNNING 状态返回envOptions等字段、结束状态返回finishedTime、errorMsg”的规则，查不到时返回{"jobId" : ""}。

返回所有已完成的作业信息

接口：GET /hazelcast/rest/maps/finished-jobs/:state

参数

响应

[ { "jobId": "", "jobName": "", "jobStatus": "", "errorMsg": null, "createTime": "", "finishTime": "", "jobDag": { "jobId": "", "envOptions": [], "vertexInfoMap": [ { "vertexId": 1, "type": "", "vertexName": "", "tablePaths": [ "" ] } ], "pipelineEdges": {} }, "metrics": "" } ]

返回系统监控信息

接口：GET /hazelcast/rest/maps/system-monitoring-information

无请求参数，返回节点级别的系统监控信息，涵盖 CPU、物理内存、堆内存、GC、线程数、Hazelcast 内部执行队列与连接数等指标，可用于构建节点健康度看板或采集进 Prometheus 等监控系统。

响应

[ { "isMaster": "true", "host": "localhost", "port": "5801", "processors":"8", "physical.memory.total":"16.0G", "physical.memory.free":"16.3M", "swap.space.total":"0", "swap.space.free":"0", "heap.memory.used":"135.7M", "heap.memory.free":"440.8M", "heap.memory.total":"576.5M", "heap.memory.max":"3.6G", "heap.memory.used/total":"23.54%", "heap.memory.used/max":"3.73%", "minor.gc.count":"6", "minor.gc.time":"110ms", "major.gc.count":"2", "major.gc.time":"73ms", "load.process":"24.78%", "load.system":"60.00%", "load.systemAverage":"2.07", "thread.count":"117", "thread.peakCount":"118", "cluster.timeDiff":"0", "event.q.size":"0", "executor.q.async.size":"0", "executor.q.client.size":"0", "executor.q.client.query.size":"0", "executor.q.client.blocking.size":"0", "executor.q.query.size":"0", "executor.q.scheduled.size":"0", "executor.q.io.size":"0", "executor.q.system.size":"0", "executor.q.operations.size":"0", "executor.q.priorityOperation.size":"0", "operations.completed.count":"10", "executor.q.mapLoad.size":"0", "executor.q.mapLoadAllKeys.size":"0", "executor.q.cluster.size":"0", "executor.q.response.size":"0", "operations.running.count":"0", "operations.pending.invocations.percentage":"0.00%", "operations.pending.invocations.count":"0", "proxy.count":"8", "clientEndpoint.count":"0", "connection.active.count":"2", "client.connection.count":"0", "connection.count":"0" } ]

提交作业

接口：POST /hazelcast/rest/maps/submit-job

如果作业提交成功，返回 jobId 和 jobName。

参数

请求体

请求体为标准的 SeaTunnel 作业配置（JSON 形式，等价于env/source/transform/sink四段式配置的 JSON 表达）：

{ "env": { "job.mode": "batch" }, "source": [ { "plugin_name": "FakeSource", "plugin_output": "fake", "row.num": 100, "schema": { "fields": { "name": "string", "age": "int", "card": "int" } } } ], "transform": [ ], "sink": [ { "plugin_name": "Console", "plugin_input": ["fake"] } ] }

响应

{ "jobId": 733584788375666689, "jobName": "rest_api_test" }

源码侧的入口为 SubmitJobServlet.java 与 RestJobExecutionEnvironment.java，后者负责将 HTTP 请求参数转换为引擎内部的作业提交环境。相关行为（含isStartWithSavePoint从 SavePoint 恢复、配置加密解密等）在测试用例 RestApiSubmitJobStartWithSavePointTest.java 与 RestApiSubmitJobConfigShadeDecryptTest.java 中有端到端验证。

批量提交作业

接口：POST /hazelcast/rest/maps/submit-jobs

如果作业提交成功，返回每个作业的 jobId 和 jobName。

参数（在请求体中 params 字段中添加）

请求体

请求体为作业配置数组，每个元素的params字段携带该作业的参数：

[ { "params":{ "jobId":"123456", "jobName":"SeaTunnel-01" }, "env": { "job.mode": "batch" }, "source": [ { "plugin_name": "FakeSource", "plugin_output": "fake", "row.num": 1000, "schema": { "fields": { "name": "string", "age": "int", "card": "int" } } } ], "transform": [ ], "sink": [ { "plugin_name": "Console", "plugin_input": ["fake"] } ] }, { "params":{ "jobId":"1234567", "jobName":"SeaTunnel-02" }, "env": { "job.mode": "batch" }, "source": [ { "plugin_name": "FakeSource", "plugin_output": "fake", "row.num": 1000, "schema": { "fields": { "name": "string", "age": "int", "card": "int" } } } ], "transform": [ ], "sink": [ { "plugin_name": "Console", "plugin_input": ["fake"] } ] } ]

响应

[ { "jobId": "123456", "jobName": "SeaTunnel-01" },{ "jobId": "1234567", "jobName": "SeaTunnel-02" } ]

停止作业

接口：POST /hazelcast/rest/maps/stop-job

如果作业成功停止，返回 jobId。

参数

请求体

{ "jobId": 733584788375666689, "isStopWithSavePoint": false, "force": false }

响应

{ "jobId": 733584788375666689 }

注意事项：

• 如果作业状态为 DOING_SAVEPOINT 且保存点未成功完成，在启用 force 选项时执行的强制停止操作会将作业状态设置为 CANCELED。
• 强制停止可能导致检查点数据不完整或处于不一致状态，仅应在异常或非正常情况下使用。

批量停止作业

接口：POST /hazelcast/rest/maps/stop-jobs

请求体为停止参数的数组，字段与单个停止作业接口一致（jobId、isStopWithSavePoint、force）。

请求体

[ { "jobId": 881432421482889220, "isStopWithSavePoint": false, "force": false }, { "jobId": 881432456517910529, "isStopWithSavePoint": false, "force": false } ]

响应

[ { "jobId": 881432421482889220 }, { "jobId": 881432456517910529 } ]

加密配置

接口：POST /hazelcast/rest/maps/encrypt-config

如果配置加密成功，则返回加密后的配置。有关自定义加密的更多信息，请参阅 配置-加密-解密 文档。该接口适合在提交作业前对配置中的账号、密码等敏感字段做统一加密（如示例中的shade.identifier: base64），加密逻辑由 EncryptConfigService.java 实现。

请求体

{ "env": { "parallelism": 1, "shade.identifier":"base64" }, "source": [ { "plugin_name": "MySQL-CDC", "schema" : { "fields": { "name": "string", "age": "int" } }, "plugin_output": "fake", "parallelism": 1, "hostname": "127.0.0.1", "username": "seatunnel", "password": "seatunnel_password", "table-name": "inventory_vwyw0n" } ], "transform": [ ], "sink": [ { "plugin_name": "Clickhouse", "host": "localhost:8123", "database": "default", "table": "fake_all", "username": "seatunnel", "password": "seatunnel_password" } ] }

响应

可见 source 与 sink 中的username、password字段已被加密为 Base64 字符串：

{ "env": { "parallelism": 1, "shade.identifier": "base64" }, "source": [ { "plugin_name": "MySQL-CDC", "schema": { "fields": { "name": "string", "age": "int" } }, "plugin_output": "fake", "parallelism": 1, "hostname": "127.0.0.1", "username": "c2VhdHVubmVs", "password": "c2VhdHVubmVsX3Bhc3N3b3Jk", "table-name": "inventory_vwyw0n" } ], "transform": [], "sink": [ { "plugin_name": "Clickhouse", "host": "localhost:8123", "database": "default", "table": "fake_all", "username": "c2VhdHVubmVs", "password": "c2VhdHVubmVsX3Bhc3N3b3Jk" } ] }

更新运行节点的 tags

接口：POST /hazelcast/rest/maps/update-tags

因为更新只能针对某个节点，因此需要用当前节点的ip:port访问对应节点来发起更新。请求成功时返回success信息。

更新节点 tags

如果请求参数是Map对象，表示要更新当前节点的 tags。

请求体：

{ "tag1": "dev_1", "tag2": "dev_2" }

响应：

{ "status": "success", "message": "update node tags done." }

移除节点 tags

如果参数为空Map对象，表示要清除当前节点的 tags。

请求体：

{}

响应：

{ "status": "success", "message": "update node tags done." }

请求参数异常

• 如果请求参数为空：

{ "status": "fail", "message": "Request body is empty." }

• 如果参数不是Map对象：

{ "status": "fail", "message": "Invalid JSON format in request body." }

节点 tags 与集群标签过滤体系配合使用：通过update-tags给节点打标后，即可在/overview?tag1=value1等接口中按标签过滤节点指标，实现简单的资源分组与亲和性运维。

获取所有节点日志内容

接口：GET /hazelcast/rest/maps/logs/:jobId

当jobId为空时，返回所有节点的日志信息；否则返回指定jobId在所有节点的日志列表。

返回所有日志文件列表

如果你想先查看日志列表，可以通过GET请求获取日志列表：http://localhost:5801/hazelcast/rest/maps/logs?format=json

[ { "node": "localhost:5801", "logLink": "http://localhost:5801/hazelcast/rest/maps/logs/job-899485770241277953.log", "logName": "job-899485770241277953.log" }, { "node": "localhost:5801", "logLink": "http://localhost:5801/hazelcast/rest/maps/logs/job-899470314109468673.log", "logName": "job-899470314109468673.log" } ]

当前支持的格式有json和html，默认为html。

例子

• 获取所有节点 jobId 为733584788375666689的日志信息：http://localhost:5801/hazelcast/rest/maps/logs/733584788375666689
• 获取所有节点日志列表：http://localhost:5801/hazelcast/rest/maps/logs
• 获取所有节点日志列表以 JSON 格式返回：http://localhost:5801/hazelcast/rest/maps/logs?format=json
• 获取日志文件内容：http://localhost:5801/hazelcast/rest/maps/logs/job-898380162133917698.log

获取单节点日志内容

接口：GET /hazelcast/rest/maps/log

与上一节的全节点接口不同，/log只返回接收请求的当前节点的日志列表，由 CurrentNodeLogServlet.java 处理。

例子

• 获取当前节点的日志列表：http://localhost:5801/hazelcast/rest/maps/log
• 获取日志文件内容：http://localhost:5801/hazelcast/rest/maps/log/job-898380162133917698.log

典型使用场景小结

1. 作业自动化提交：通过POST /submit-jobs以编程方式批量投递作业，再轮询GET /job-info/:jobId跟踪状态，配合metrics中的SourceReceivedQPS、SinkCommittedQPS等字段做实时吞吐监控。
2. 可视化监控看板：使用GET /overview展示集群 Slot 与作业计数，GET /running-jobs获取作业 DAG 渲染拓扑图，GET /system-monitoring-information与GET /resource/workers展示节点健康度。
3. 故障诊断：作业卡死时调用GET /thread-dump导出线程堆栈，GET /logs/:jobId跨节点拉取日志，结合GET /finished-jobs/FAILED查看历史失败作业与errorMsg。
4. 多环境运维：通过POST /update-tags为节点打上环境标签，再用带 tag 的/overview查询实现按标签维度查看集群指标。
5. 安全合规：提交前调用POST /encrypt-config加密配置中的敏感字段，避免明文密码进入作业配置文件。

以上所有接口均要求先在hazelcast.yaml中将rest-api.enabled设为true，且服务随节点进程启动后立即可用，无需额外部署组件。由于 V1 已弃用，新系统建议评估迁移至 REST API V2，而本文所覆盖的作业生命周期管理细节可进一步参考 REST API 作业生命周期文档。

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel