Apache DolphinScheduler GRPC 任务节点实战参数配置、状态码校验与底层调用原理【免费下载链接】dolphinschedulerApache DolphinScheduler is the modern data orchestration platform. Agile to create high performance workflow with low-code项目地址: https://gitcode.com/GitHub_Trending/dol/dolphinscheduler本文以 Apache DolphinScheduler 的 GRPC 任务节点为核心系统讲解如何在数据编排工作流中直接调用 gRPC 服务从创建任务、配置请求地址与 Protobuf 服务定义到校验 gRPC 状态码、在下游任务中引用返回值并深入源码剖析其动态 Protobuf 解析与调用实现原理。读完本文你将掌握 GRPC 任务节点的全部配置项含义、参数替换规则与输出参数用法能够独立将任意 gRPC 服务接入 DolphinScheduler 工作流。综述GRPC 节点能做什么GRPC 任务节点用于执行 gRPC 类型的任务核心能力包括直接调用任意 gRPC 服务无需预先生成服务端/客户端代码只需在任务中粘贴 Protobuf 服务定义即可动态发起 RPC 调用支持 gRPC 状态码校验默认校验响应状态码为OK也支持自定义期望状态码做精确匹配支持 SSL/TLS 安全传输凭证类型支持不安全INSECURE与客户端默认 SSL/TLS 两种模式支持参数注入与结果输出任务参数支持内置参数替换调用结果以response输出参数形式暴露给下游任务引用。该节点的插件实现位于 dolphinscheduler-task-plugin/dolphinscheduler-task-grpc核心类包括 GrpcTask.java任务执行与校验、GrpcParameters.java参数模型以及 GrpcDynamicService.java动态调用引擎。创建 GRPC 任务在 DolphinScheduler 中创建 GRPC 任务的操作路径为点击项目管理 - 项目名称 - 工作流定义点击创建工作流按钮进入 DAG 编辑页面从工具栏拖动GRPC任务节点标识为 GRPC 的任务图标到画板中双击节点打开任务配置表单填写下方任务参数中的各项配置配置完成后保存任务即可作为工作流中的一个节点参与调度执行。GRPC 节点的插件通过AutoService(TaskChannelFactory.class)机制注册插件名称固定为GRPC见 GrpcTaskChannelFactory.java。任务参数详解GRPC 任务参数默认参数说明请参考 DolphinScheduler任务参数附录 的默认任务参数一栏。GRPC 节点专属参数如下任务参数描述请求地址gRPC 请求 URL需使用hostname:port格式例如localhost:50051gRPC 凭证类型支持None不安全、客户端默认 SSL/TLS两种凭证类型Protobuf 服务定义用于.proto文件中定义服务的 protobuf 代码保存时会将该内容转换为 JSON Descriptor请求方法要调用的 rpc 方法需在服务定义中定义写作Greeter/SayHello格式服务名/方法名消息内容使用 JSON 定义的请求消息请求时会合并到服务定义中校验条件支持默认 gRPC 状态码OK、自定义状态码校验内容当校验条件选择自定义响应码时需填写校验内容需与 gRPC 官方状态码定义一致自定义参数是 gRPC 局部的用户自定义参数会替换脚本中以${变量}的内容从源码角度看这些字段在 GrpcParameters.java 中一一对应其中几个关键字段存在默认值与前置校验逻辑channelCredentialType默认值为INSECURE对应枚举 GrpcCredentialType.java 中的INSECURE与TLS_DEFAULT客户端默认凭据grpcCheckCondition默认值为STATUS_CODE_DEFAULT对应枚举 GrpcCheckCondition.java 中的默认状态码OK与自定义状态码两种模式connectTimeoutMs连接超时单位毫秒为任务必填的隐含条件checkParameters()方法要求请求地址非空且连接超时大于 0否则任务参数校验不通过。checkParameters()还会在参数校验阶段做一次预编译式检查将grpcServiceDefinitionJSON反序列化为 protobufFileDescriptor并尝试把方法名与 JSON 消息合并到服务定义中任何一步失败如方法名不存在、JSON 消息与消息类型不匹配都会导致参数校验失败从而在任务启动前拦截错误配置。对应实现可查看 GrpcParameters.java。关于 Protobuf 服务定义表单中填写的 protobuf 代码在保存时会被转换为JSON Descriptor存储对应参数字段grpcServiceDefinitionJSON。转换链路为JSONDescriptorHelper.java 将 JSON 解析为Root映射对象再由 JSONDescriptorParser.java 构建标准的 protobufFileDescriptor。Root模型见 mapping/Root.java完整映射了 proto 文件的 namespace、service、method、message 字段、枚举、oneof、map 等结构。一个典型的 proto3 服务定义示例如下syntax proto3; // The greeting service definition. service Greeter { rpc SayHello (HelloRequest) returns (HelloReply); rpc SayHelloAgain (HelloRequest) returns (HelloReply); } // The request message containing the users name message HelloRequest { string name 1; } // The response message containing the greetings message HelloReply { string message 1; }关于请求方法格式请求方法必须写作服务名/方法名格式例如Greeter/SayHello。底层解析逻辑位于 GrpcDynamicService.javaMethodName内部类以/GrpcConstants.SERVICE_METHOD_SEPERATOR分割字符串恰好拆分为 serviceName 与 rpcName 两段随后通过fileDescriptor.findServiceByName(...)与pServiceDescriptor.findMethodByName(...)在服务定义中查找对应方法若找不到会抛出明确的GrpcParserException异常信息中会附带该服务实际拥有的方法列表便于排查。关于消息内容消息内容使用 JSON 定义请求发起前会合并merge到 protobuf 消息中。合并过程使用JsonFormat.parser().ignoringUnknownFields().merge(messageJSON, requestBuilder)见 GrpcDynamicService.java这意味着 JSON 中可以包含服务定义中不存在的字段多余字段会被忽略但字段名需与 proto 定义中的消息字段对应。任务输出参数任务参数描述responseVARCHAR符合 ProtoJS 格式gRPC 请求的返回结果GRPC 任务执行成功后会将服务端返回的响应消息序列化为 JSON 字符串注册为任务输出参数response。可以在下游任务中使用${taskName.response}引用任务输出参数。例如当前task1为 GRPC 任务下游任务可以使用${task1.response}引用 task1 的输出参数。从实现看输出参数的写入逻辑在 GrpcTask.java 的addDefaultOutput方法中返回消息经由JsonFormat.printer().omittingInsignificantWhitespace()压缩空白后打印为 JSON 字符串并以${taskName}.response作为属性名、VARCHAR作为数据类型Direct.OUT写入任务的值池val pool供工作流后续节点解析替换。任务样例以下为一个完整的 GRPC 任务配置样例配置项均支持通过内置参数替换请求地址访问目标 gRPC 服务的地址这里为本地的localhost:5005150051 端口gRPC 凭证类型不安全Protobuf 定义gRPC 服务所使用的 protobuf 定义见上文示例方法名称要调用的 rpc 方法Greeter/SayHello格式消息内容使用 JSON 定义的请求消息例如{name: my name}校验条件默认 gRPC 状态码 OK 或自定义状态码校验内容校验条件为自定义状态码时填写内容为精确匹配 gRPC 状态码字符串如NOT_FOUND、UNAVAILABLE等。节点配置界面如图从第一张截图可以看到示例中在Protobuf 定义文本框内粘贴了Greeter服务的完整 proto 代码含SayHello/SayHelloAgain两个方法方法名称填写Greeter/SayHello消息内容填写 JSON{name: my name}。第二张截图展示了高级配置区校验条件选择默认状态码OK连接超时设置为60000毫秒60 秒并可通过自定义参数区域动态添加局部变量参与任务参数替换。连接超时的作用连接超时connectTimeoutMs会作为 gRPC 调用的deadline截止时间传入当超时值大于 0 时CallOptions.DEFAULT.withDeadlineAfter(timeout, TimeUnit.MILLISECONDS)为调用设置超时上限见 GrpcDynamicService.java超过该时间仍未返回时gRPC 会抛出StatusRuntimeException(DEADLINE_EXCEEDED)任务据此判定失败。由于参数校验要求该值必须大于 0因此配置任务时务必显式填写合理的超时时间。底层调用原理从配置到一次 RPC 调用结合源码GRPC 任务从初始化到执行完成的完整链路如下1. 任务初始化与参数校验GrpcTask.init() 将任务参数字符串解析为GrpcParameters并调用checkParameters()完成前置校验JSON Descriptor 合法性、方法名与消息可合并性、请求地址非空、超时大于 0校验失败会抛出GrpcTaskException任务直接失败。2. 创建 gRPC ChannelGrpcTask.handle() 根据凭证类型选择通道创建方式凭证类型为INSECURE时使用InsecureChannelCredentials.create()凭证类型为TLS_DEFAULT时使用TlsChannelCredentials.create()创建默认的客户端 TLS 凭据实现 SSL/TLS 加密传输。两种方式最终都通过NettyChannelBuilder.forTarget(targetAddr, channelCredentials)构建 Netty 驱动的ManagedChannel见 GrpcDynamicService.java。3. 动态方法调用通道就绪后GrpcDynamicService.call(...)会完成一次动态无预编译 stub的单向调用解析服务名/方法名从FileDescriptor中定位 Service 与 Method通过ProtoUtils.marshaller(DynamicMessage.getDefaultInstance(...))为请求/响应构造 marshaller组装出MethodDescriptor根据 proto 定义中isServerStreaming/isClientStreaming标志自动识别调用类型UNARY、SERVER_STREAMING、CLIENT_STREAMING、BIDI_STREAMING见 GrpcDynamicService.java将 JSON 消息合并进DynamicMessage请求构建器通过ClientCalls.blockingUnaryCall(...)发起阻塞式调用返回DynamicMessage响应。需要说明的是虽然代码中根据 proto 定义计算出了流式调用类型但实际调用固定走blockingUnaryCall阻塞一元调用即当前版本适用于unary 风格的单请求-单响应 RPC场景。4. 状态码校验与任务判定调用结束后GrpcTask.validateResponse() 按校验条件判定任务成败STATUS_CODE_DEFAULT默认状态码 OK检查 gRPC 状态statusCode.isOk()非 OK 即任务失败STATUS_CODE_CUSTOM自定义状态码将校验内容字符串通过Status.Code.valueOf(condition)转换为标准状态码枚举再与调用返回的状态码做精确匹配statusCode ! expectedCode即失败。若填写的校验内容不是合法状态码会抛出GrpcTaskException。此外调用过程中的异常被包装为StatusRuntimeException捕获后同样进入状态码校验流程因此当校验条件为自定义状态码时即使远程服务返回错误状态如NOT_FOUND只要与期望状态码一致任务也会被判定为成功——这一设计可用于预期失败类的业务校验场景。5. 结果输出与下游引用成功路径上响应消息打印为压缩 JSON 后通过addDefaultOutput写入值池下游任务即可通过${taskName.response}引用。测试验证仓库中提供了完整的单元测试 GrpcTaskTest.java测试使用 mock 的 gRPC 服务端基于InsecureServerCredentials启动本地 Server通过TaskTesterGrpc等预生成的桩代码模拟testOK、testFail等 RPC 方法分别验证任务成功、StatusRuntimeException失败、输出参数写入Property的prop/direct/type/value字段断言等行为。此外 GrpcParserTest.java 与 GrpcParametersTest.java 分别覆盖了 proto/JSON Descriptor 解析与参数校验逻辑可作为理解插件行为的参考。使用建议与注意事项请求地址必须为hostname:portURL 中的端口将用于 gRPC 通道寻址务必确保 Worker 节点网络可访问该地址方法名格式不可省略服务名Greeter/SayHello中的服务名与 proto 中的service块必须严格一致区分大小写消息字段名需与 proto 对齐虽然合并时ignoringUnknownFields会忽略多余字段但目标字段名与类型不匹配会导致合并失败、任务启动即报错自定义校验内容需为合法状态码填写时应使用 gRPC 官方状态码枚举名如OK、NOT_FOUND、UNAVAILABLE、DEADLINE_EXCEEDED等精确匹配、不区分大小写规则以Status.Code.valueOf为准连接超时必填且应合理设置超时既作为参数校验的硬性要求也直接决定调用 deadline建议按实际服务响应耗时设置TLS 场景注意证书环境TLS_DEFAULT使用 JVM 默认信任库加载的客户端凭据服务端证书需在 Worker 节点可信范围内否则会握手失败。GRPC 任务节点让 DolphinScheduler 无需额外开发即可编排调用微服务生态中的 gRPC 接口配合输出参数response与内置参数替换能力可作为微服务间数据编排与任务联动的高效补充节点。【免费下载链接】dolphinschedulerApache DolphinScheduler is the modern data orchestration platform. Agile to create high performance workflow with low-code项目地址: https://gitcode.com/GitHub_Trending/dol/dolphinscheduler创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考