Conductor JSON JQ Transform 任务完全指南：用 jq 表达式在 Workflow 中转换与过滤 JSON 数据

Conductor JSON JQ Transform 任务完全指南用 jq 表达式在 Workflow 中转换与过滤 JSON 数据【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductorJSON JQ Transform 是 Conductor 内置的同步系统任务系统任务类型JSON_JQ_TRANSFORM它借助 jq 表达式在 Workflow 内部直接完成 JSON 数据的转换、筛选与重排常用于把一个任务的输出加工成另一个任务的输入。读完本文你将掌握该任务的参数约定、输出结构、实战配置写法以及底层实现与异常处理机制能够在自己的 Workflow 中直接落地使用。任务概述在编排式工作流中任务之间往往需要传递结构复杂的 JSON 数据而下游任务通常只关心其中的一小部分字段。JSON JQ Transform 任务正是为解决这一痛点而设计它把 jq 过滤器作为胶水让开发者无需编写自定义 Worker即可在工作流定义中声明式地完成数据变换。在 Workflow 的任务定义中将type字段设置为JSON_JQ_TRANSFORM即可启用该任务type : JSON_JQ_TRANSFORM在 Conductor 源码中该任务的实现位于 json-jq-task 模块任务核心类 JsonJqTransform.java 以Component(JSON_JQ_TRANSFORM)注册继承自WorkflowSystemTask并在start()方法中完成全部转换逻辑JsonJqTransform.java。任务的类型常量定义于 TaskType.java任务建模把工作流定义中的输入参数解析为任务实际输入则由 JsonJQTransformTaskMapper.java 负责。任务参数JSON JQ Transform 的全部参数都声明在任务的inputParameters中核心参数如下参数类型说明必填/可选queryExpressionString用于转换 JSON 数据的 jq 过滤器表达式。jq 官方文档与手册提供了完整的过滤器构造语法也可在 jqplay 上交互式测试表达式。必填inputParametersMap[String, Any]供 jq 表达式求值使用的输入数据集合。必填其中queryExpression与其他输入值一并放置在inputParameters中例如persons、key1等业务字段会作为 jq 求值时的顶层上下文变量供表达式引用。从实现上看任务在执行时会读取输入中的queryExpression字符串随后将整个inputParametersMap序列化为 JSON 根节点交给 jq 求值JsonJqTransform.java。因此表达式里写.persons、.key1.value1这类路径实际访问的就是inputParameters中对应键的值。参数校验行为源码还揭示了一个易踩坑点如果inputParameters中缺少queryExpression任务不会执行转换而是直接以失败告终reasonForIncompletion会被设置为Missing queryExpression in input parametersJsonJqTransform.java。对应地单元测试 JsonJqTransformTest.java 验证了当表达式语法非法如只写一个{时任务输出中的error字段会包含 jq 编译器的报错信息。输出结构任务执行完成后会在任务输出中返回以下三个参数名称类型说明resultList[Map[String, Any]]jq 过滤器返回的resultList中的第一个元素。resultListList[List[Map[String, Any]]]jq 过滤器返回的全部结果列表。errorString可选字段当 jq 过滤器执行失败时给出错误信息。结合源码可以更精确地理解这两个结果字段jq 的求值结果是ListJsonNoderesultList保存的是转换后的完整列表result取列表中的第一个元素若列表为空result为null若 jq 返回null则result与resultList均为nullJsonJqTransform.java输出中的标量类型会根据 JSON 节点类型做转换对象转为Map、数组转为List、布尔转为Boolean、整数转为Long、浮点转为Double、字符串与其余类型转为StringJsonJqTransform.java。因此result实际可能是对象、数组、字符串、数字等任意 JSON 类型单元测试中分别对 Map 结果、List 结果、String 结果与 null 结果做了覆盖验证。表达式求值失败时任务状态置为FAILEDreasonForIncompletion与error输出会包含提取到的第一条有效异常消息源码会逐层遍历异常链并过滤掉内容为 N/A 的节点见 JsonJqTransform.java 与 extractFirstValidMessage。与下游任务的衔接result通常会被下游任务引用。例如在决策任务DECISION中可以写作${json_jq_1.output.result}作为caseValueParam的取值来源详见后文与决策任务联动一节的完整工作流示例。标准配置示例以下是一个完整的 JSON JQ Transform 任务定义persons数组作为输入queryExpression把每个用户映射为{user:{email,id}}结构{ name: json_transform, taskReferenceName: json_transform_ref, type: JSON_JQ_TRANSFORM, inputParameters: { persons: [ { name: some, last: name, email: mailmail.com, id: 1 }, { name: some2, last: name2, email: mail2mail.com, id: 2 } ], queryExpression: .persons | map({user:{email,id}}) } }除name、taskReferenceName、type、inputParameters外任务定义还支持decisionCases、defaultCase、forkTasks、startDelay、joinOn、optional、defaultExclusiveJoinTask、asyncComplete、loopOver等通用字段在真实的工作流 JSON 中这些字段通常会完整写出见 simple_json_jq_transform_integration_test.json。实战示例简单示例数组合并下面的任务把key1.value1与key2.value2两个字符串数组通过 jq 的运算符拼接为key3数组{ name: jq_example_task, taskReferenceName: my_jq_example_task, type: JSON_JQ_TRANSFORM, inputParameters: { key1: { value1: [ a, b ] }, key2: { value2: [ c, d ] }, queryExpression: { key3: (.key1.value1 .key2.value2) } } }该任务输出如下。由于 jq 只产生一个结果此时resultList与result内容相同{ result: { key3: [ a, b, c, d ] }, resultList: [ { key3: [ a, b, c, d ] } ] }集成测试 JsonJQTransformSpec.groovy 中也有等价场景工作流输入in1.array [a,b]、in2.array [c,d]经表达式.input as $_ | { out: ($_.in1.array $_.in2.array) }转换为out: [a,b,c,d]。注意该表达式使用了变量绑定as $_因为输入整体被包在input键下input: ${workflow.input}。简化稠密数据从 GitHub API 响应中提取信息真实场景中HTTP 任务返回的第三方 API 响应往往字段冗长。下面的例子演示如何用 JSON JQ Transform 精简 GitHub stargazers 接口的响应——该接口对单个用户的返回形如body:[ { starred_at:2016-12-14T19:55:46Z, user:{ login:lzehrung, id:924226, node_id:MDQ6VXNlcjkyNDIyNg, avatar_url:https://avatars.githubusercontent.com/u/924226?v4, gravatar_id:, url:https://api.github.com/users/lzehrung, html_url:https://github.com/lzehrung, followers_url:https://api.github.com/users/lzehrung/followers, following_url:https://api.github.com/users/lzehrung/following{/other_user}, gists_url:https://api.github.com/users/lzehrung/gists{/gist_id}, starred_url:https://api.github.com/users/lzehrung/starred{/owner}{/repo}, subscriptions_url:https://api.github.com/users/lzehrung/subscriptions, organizations_url:https://api.github.com/users/lzehrung/orgs, repos_url:https://api.github.com/users/lzehrung/repos, events_url:https://api.github.com/users/lzehrung/events{/privacy}, received_events_url:https://api.github.com/users/lzehrung/received_events, type:User, site_admin:false } } ]实际业务只需要在指定日期之后 star 过仓库的用户的starred_at与login两个字段。日期由工作流输入${workflow.input.cutoff_date}提供。于是可以用如下任务配置完成精简{ name: jq_cleanup_stars, taskReferenceName: jq_cleanup_stars_ref, inputParameters: { starlist: ${hundred_stargazers_ref.output.response.body}, queryExpression: [.starlist[] | select (.starred_at \${workflow.input.cutoff_date}\) |{occurred_at:.starred_at, member: {github: .user.login}}] }, type: JSON_JQ_TRANSFORM, decisionCases: {}, defaultCase: [], forkTasks: [], startDelay: 0, joinOn: [], optional: false, defaultExclusiveJoinTask: [], asyncComplete: false, loopOver: [] }这里把 HTTP 任务的响应体body放入starlist参数queryExpression依次完成三件事用[]遍历数组、用select过滤starred_at满足日期条件的元素、用对象构造语法重排输出。最终每条数据变成{ occurred_at: date from JSON, member:{ github : github Login from JSON } }注意queryExpression整体用[]包裹表示期望输出是一个数组。这是该任务的一个重要约定——表达式的输出结构对象还是数组完全由表达式本身决定result字段也会相应呈现为对应类型。与决策任务联动一个完整的可运行工作流仓库自带的集成测试资源 json_jq_transform_result_integration_test.json 展示了一个更完整的用法JSON JQ Transform 与 DECISION 任务协同工作。其流程为json_jq_1用条件表达式if(.data | length 0) then EXISTS else CREATE end根据数据是否为空产出字符串结果decide_1以${json_jq_1.output.result}作为outcome按CREATE分支进入下一步json_jq_2用.inputData.request.transitions | map(.name)提取transitions数组中每个对象的name产出 3 个元素的字符串列表decide_2通过caseExpression判断requestedAction是否在可用动作列表中决定是否调用 HTTP 任务。对应的集成测试 JsonJQTransformSpec.groovy 完整断言了这一链路可验证字符串结果 → 决策与数组结果 → 决策两种衔接方式。用 Workflow 输入引用串联任务一个常见的编排手法是先用一个 JQ 任务把 Workflow 输入整理成默认参数集合再让后续 JQ 任务引用其中的片段。仓库测试资源 sequential_json_jq_transform_integration_test.json 即展示了这种用法第一个任务default_variables使用带默认值合并运算符//的表达式从 Workflow 输入中提取requestTransform、responseTransform、method、document、successExpression等字段缺失时回退到默认字符串如// .body第二个任务request_transform的queryExpression直接引用前一个任务的输出${default_variables.output.result.requestTransform}——也就是把上一个 JQ 任务算出来的表达式字符串再作为下一个 JQ 任务的表达式执行。该测试 JsonJQTransformSpec.groovy 验证了Beary Beariston you are the Brown Bear这样的动态拼接结果说明 JSON JQ Transform 支持表达式来自运行时输入是构建参数化数据管线的有效手段。底层实现与执行原理从 JsonJqTransform.java 可以完整还原该任务的执行链路注册与继承任务类通过 SpringComponent注册为名为JSON_JQ_TRANSFORM的系统任务 Bean继承WorkflowSystemTask因此在 Workflow 中属于同步系统任务——一旦调度即立刻在当前执行线程内完成不依赖外部 Worker 轮询。编译与缓存构造时创建全局 jq 作用域Scope.newEmptyScope()并加载内置函数表达式通过JsonQuery::compile编译。为提高性能编译后的JsonQuery存放在 Caffeine 缓存中写入 1 小时后过期最大容量 1000 条createQueryCache。这意味着重复使用相同表达式的工作流几乎不会产生重复编译开销。求值start()中把任务inputData即inputParameters整体序列化为 JSON 根节点在子作用域中执行query.apply(childScope, input)得到ListJsonNode结果。结果规整将每个JsonNode按类型转换为 Java 对象Map/List/Boolean/Long/Double/String/null写入result与resultList输出。失败处理任何编译或求值异常都会将任务标记为FAILED并通过遍历异常链提取第一条不含 N/A 的消息写入error输出与reasonForIncompletion便于在工作流状态中直接定位原因。失败场景与重跑由于queryExpression是运行时解析的表达式写错、输入结构不符都会导致任务失败。集成测试 JsonJQTransformSpec.groovy 演示了一个典型错误当in1、in2从对象变成字符串后表达式in1.array无法取下标任务以Cannot index string with string array失败。同一个测试还展示了**失败后重跑rerun**的恢复路径通过RerunWorkflowRequest指定从失败任务重跑并传入修正后的输入工作流即可恢复为COMPLETED状态JsonJQTransformSpec.groovy。最佳实践建议基于文档与源码总结几条实战建议先交互验证表达式再上线在 jqplay 等工具中针对真实输入测试queryExpression重点验证输入为null、缺字段、类型不符时的行为避免运行时才暴露语法或路径错误。用变量绑定避免重复取值当表达式需要多处引用深层路径时优先使用as $_或as $x绑定变量既提升可读性也减少路径书写错误见集成测试中的.input as $_用法。善用//缺省运算符为参数提供默认值sequential_json_jq_transform_integration_test.json展示了用//为可空输入提供兜底值的写法能显著提升任务对多样输入的鲁棒性。区分result与resultList下游若只消费单个对象/数组用${taskRef.output.result}若需要遍历 jq 产生的多条输出用${taskRef.output.resultList}。注意当 jq 表达式未产生结果时result为null下游引用需做好容错。为任务定义optional与重跑策略JQ 任务可能因上游数据形态变化而失败可在任务定义中按业务需要设置optional: true或在 Workflow 层配置失败重试/重跑流程重跑时通过taskInput修正输入。小结JSON JQ Transform 把 jq 的表达能力带进了 Conductor Workflow让数据整形、字段过滤、结构重排在流程定义内即可完成省去了编写与部署自定义 Worker 的成本。其核心要点可以概括为参数集中在inputParametersqueryExpression必填、输出固定为result/resultList/error、表达式决定输出结构、失败信息会透传到任务状态。结合 JsonJqTransform.java 的实现与 test-harness 中的集成测试你可以在仓库中进一步探索其边界行为与更多编排玩法。【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考