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 监控与运维 APIV1 版本。读完后你可以掌握如何在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: truenetwork: 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 realmSeaTunnel Web UI响应头未开启时请求直接放行。API 接口总览V1 版本提供以下接口均位于/hazelcast/rest/maps路径下接口方法功能/option-rulesGET获取 Connector 运行时完整 OptionRule 元数据/overviewGET返回 Zeta 集群概览/resource/workersGET查询已注册 Worker 的资源快照/thread-dumpGET返回当前节点的线程堆栈信息/running-jobsGET返回所有作业及其当前状态的概览/job-info/:jobIdGET返回作业的详细信息含全量指标/running-job/:jobIdGET已弃用请使用/job-info/:jobId/finished-jobs/:stateGET返回所有已完成的作业信息/system-monitoring-informationGET返回系统监控信息/submit-jobPOST提交单个作业/submit-jobsPOST批量提交作业/stop-jobPOST停止作业/stop-jobsPOST批量停止作业/encrypt-configPOST加密作业配置/update-tagsPOST更新运行节点的 tags/logs/:jobIdGET获取所有节点日志内容/logGET获取单节点日志内容以上端点常量均定义在 RestConstant.java 中且每个端点在 servlet 目录 下有对应的 Servlet 实现、在 service 目录 下有对应的 Service 逻辑可据此进一步阅读源码。获取 Connector 的 OptionRule接口GET /hazelcast/rest/maps/option-rules?typesourcepluginFakeSource返回指定 Connector 运行时完整的 OptionRule 元数据由 OptionRulesService.java 处理响应结构定义在 OptionRuleResponse.java。参数参数名称是否必传参数类型参数描述type是string插件类型当前支持source和sinkplugin是stringconnector 的 factory identifier例如FakeSource或Console响应{ 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.Listjava.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?tag1value1tag2value2参数参数名称是否必传参数类型参数描述tag键值对否字符串一组标签值通过该标签值过滤满足条件的节点信息响应{ 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是longjob id响应{ 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。指标字段说明字段说明SourceReceivedCount源端接收的行数SourceReceivedQPS源端接收速率行/秒SourceReceivedBytes源端接收的字节数SourceReceivedBytesPerSeconds源端接收速率字节/秒SinkWriteCountSink 写入尝试行数SinkWriteQPSSink 写入尝试速率行/秒SinkWriteBytesSink 写入尝试字节数SinkWriteBytesPerSecondsSink 写入尝试速率字节/秒SinkCommittedCountcheckpoint 成功后的 Sink 已提交行数SinkCommittedQPSSink 已提交速率行/秒SinkCommittedBytescheckpoint 成功后的 Sink 已提交字节数SinkCommittedBytesPerSecondsSink 已提交速率字节/秒TableSourceReceived*按表汇总的源指标键格式TableSourceReceivedXXX#表TableSinkWrite*按表汇总的 Sink 写入尝试键格式TableSinkWriteXXX#表TableSinkCommitted*按表汇总的 Sink 已提交指标键格式TableSinkCommittedXXX#表按表汇总的指标名常量与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参数参数名称是否必传参数类型参数描述stateoptionalstringfinished job status.FINISHED,CANCELED,FAILED,SAVEPOINT_DONE,UNKNOWABLE响应[ { 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。参数参数名称是否必传参数类型参数描述jobIdoptionalstringjob idjobNameoptionalstringjob nameisStartWithSavePointoptionalstringif job is started with save point请求体请求体为标准的 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 字段中添加参数名称是否必传参数类型参数描述jobIdoptionalstringjob idjobNameoptionalstringjob nameisStartWithSavePointoptionalstringif job is started with save point请求体请求体为作业配置数组每个元素的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。参数参数名称是否必传参数类型参数描述jobIdyeslong作业 IDisStopWithSavePointnoboolean是否通过 savepoint 方式停止作业forcenoboolean是否强制停止作业忽略 isStopWithSavePoint 参数请求体{ 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?tag1value1等接口中按标签过滤节点指标实现简单的资源分组与亲和性运维。获取所有节点日志内容接口GET /hazelcast/rest/maps/logs/:jobId参数名称是否必传参数类型参数描述jobIdoptionalstringjob id当jobId为空时返回所有节点的日志信息否则返回指定jobId在所有节点的日志列表。返回所有日志文件列表如果你想先查看日志列表可以通过GET请求获取日志列表http://localhost:5801/hazelcast/rest/maps/logs?formatjson[ { 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?formatjson获取日志文件内容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典型使用场景小结作业自动化提交通过POST /submit-jobs以编程方式批量投递作业再轮询GET /job-info/:jobId跟踪状态配合metrics中的SourceReceivedQPS、SinkCommittedQPS等字段做实时吞吐监控。可视化监控看板使用GET /overview展示集群 Slot 与作业计数GET /running-jobs获取作业 DAG 渲染拓扑图GET /system-monitoring-information与GET /resource/workers展示节点健康度。故障诊断作业卡死时调用GET /thread-dump导出线程堆栈GET /logs/:jobId跨节点拉取日志结合GET /finished-jobs/FAILED查看历史失败作业与errorMsg。多环境运维通过POST /update-tags为节点打上环境标签再用带 tag 的/overview查询实现按标签维度查看集群指标。安全合规提交前调用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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
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 监控与运维 APIV1 版本。读完后你可以掌握如何在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: truenetwork: 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 realmSeaTunnel Web UI响应头未开启时请求直接放行。API 接口总览V1 版本提供以下接口均位于/hazelcast/rest/maps路径下接口方法功能/option-rulesGET获取 Connector 运行时完整 OptionRule 元数据/overviewGET返回 Zeta 集群概览/resource/workersGET查询已注册 Worker 的资源快照/thread-dumpGET返回当前节点的线程堆栈信息/running-jobsGET返回所有作业及其当前状态的概览/job-info/:jobIdGET返回作业的详细信息含全量指标/running-job/:jobIdGET已弃用请使用/job-info/:jobId/finished-jobs/:stateGET返回所有已完成的作业信息/system-monitoring-informationGET返回系统监控信息/submit-jobPOST提交单个作业/submit-jobsPOST批量提交作业/stop-jobPOST停止作业/stop-jobsPOST批量停止作业/encrypt-configPOST加密作业配置/update-tagsPOST更新运行节点的 tags/logs/:jobIdGET获取所有节点日志内容/logGET获取单节点日志内容以上端点常量均定义在 RestConstant.java 中且每个端点在 servlet 目录 下有对应的 Servlet 实现、在 service 目录 下有对应的 Service 逻辑可据此进一步阅读源码。获取 Connector 的 OptionRule接口GET /hazelcast/rest/maps/option-rules?typesourcepluginFakeSource返回指定 Connector 运行时完整的 OptionRule 元数据由 OptionRulesService.java 处理响应结构定义在 OptionRuleResponse.java。参数参数名称是否必传参数类型参数描述type是string插件类型当前支持source和sinkplugin是stringconnector 的 factory identifier例如FakeSource或Console响应{ 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.Listjava.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?tag1value1tag2value2参数参数名称是否必传参数类型参数描述tag键值对否字符串一组标签值通过该标签值过滤满足条件的节点信息响应{ 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是longjob id响应{ 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。指标字段说明字段说明SourceReceivedCount源端接收的行数SourceReceivedQPS源端接收速率行/秒SourceReceivedBytes源端接收的字节数SourceReceivedBytesPerSeconds源端接收速率字节/秒SinkWriteCountSink 写入尝试行数SinkWriteQPSSink 写入尝试速率行/秒SinkWriteBytesSink 写入尝试字节数SinkWriteBytesPerSecondsSink 写入尝试速率字节/秒SinkCommittedCountcheckpoint 成功后的 Sink 已提交行数SinkCommittedQPSSink 已提交速率行/秒SinkCommittedBytescheckpoint 成功后的 Sink 已提交字节数SinkCommittedBytesPerSecondsSink 已提交速率字节/秒TableSourceReceived*按表汇总的源指标键格式TableSourceReceivedXXX#表TableSinkWrite*按表汇总的 Sink 写入尝试键格式TableSinkWriteXXX#表TableSinkCommitted*按表汇总的 Sink 已提交指标键格式TableSinkCommittedXXX#表按表汇总的指标名常量与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参数参数名称是否必传参数类型参数描述stateoptionalstringfinished job status.FINISHED,CANCELED,FAILED,SAVEPOINT_DONE,UNKNOWABLE响应[ { 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。参数参数名称是否必传参数类型参数描述jobIdoptionalstringjob idjobNameoptionalstringjob nameisStartWithSavePointoptionalstringif job is started with save point请求体请求体为标准的 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 字段中添加参数名称是否必传参数类型参数描述jobIdoptionalstringjob idjobNameoptionalstringjob nameisStartWithSavePointoptionalstringif job is started with save point请求体请求体为作业配置数组每个元素的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。参数参数名称是否必传参数类型参数描述jobIdyeslong作业 IDisStopWithSavePointnoboolean是否通过 savepoint 方式停止作业forcenoboolean是否强制停止作业忽略 isStopWithSavePoint 参数请求体{ 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?tag1value1等接口中按标签过滤节点指标实现简单的资源分组与亲和性运维。获取所有节点日志内容接口GET /hazelcast/rest/maps/logs/:jobId参数名称是否必传参数类型参数描述jobIdoptionalstringjob id当jobId为空时返回所有节点的日志信息否则返回指定jobId在所有节点的日志列表。返回所有日志文件列表如果你想先查看日志列表可以通过GET请求获取日志列表http://localhost:5801/hazelcast/rest/maps/logs?formatjson[ { 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?formatjson获取日志文件内容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典型使用场景小结作业自动化提交通过POST /submit-jobs以编程方式批量投递作业再轮询GET /job-info/:jobId跟踪状态配合metrics中的SourceReceivedQPS、SinkCommittedQPS等字段做实时吞吐监控。可视化监控看板使用GET /overview展示集群 Slot 与作业计数GET /running-jobs获取作业 DAG 渲染拓扑图GET /system-monitoring-information与GET /resource/workers展示节点健康度。故障诊断作业卡死时调用GET /thread-dump导出线程堆栈GET /logs/:jobId跨节点拉取日志结合GET /finished-jobs/FAILED查看历史失败作业与errorMsg。多环境运维通过POST /update-tags为节点打上环境标签再用带 tag 的/overview查询实现按标签维度查看集群指标。安全合规提交前调用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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考