di-agent-flow-datastage

使用 SDK 编写和编辑 DataStage 流

路由由其他文件负责。 请求应在此处还是在 pyflow 中编写，以及应该创建还是编辑，都由 di-agent-flow-lifecycle 技能（AUTHOR 状态）决定。本文是 SDK 参考文档，介绍其使用方式。在处理流之前，必须全面了解 di-agent-flow-lifecycle 技能。加载生命周期技能之前，不要编写任何 SDK 代码。

DataStage 批处理流以 SDK 风格的 Python 代码编写，并通过 update_datastage_flow（就地编辑现有流）或 create_datastage_flow（创建新流）提交。身份验证、项目上下文和持久化均由系统自动处理。

SDK 可以访问引擎的全部阶段目录，包括 pyflow 没有表单的所有阶段。这种覆盖能力也有代价：它需要准确的阶段类型、属性名称和枚举值，因此路由器将其视为备用方案，而不是默认选择。

这两个工具都会在调用过程中编译流。 如果编译失败，结果会返回 status: "compile_error"，并包含 flow_id、flow_link 和逐阶段的 compile_errors 列表——流仍会保存。使用同一个 flow_id 调用 update_datastage_flow 修复。绝不要创建新流，也不要删除有问题的流；编辑时要使用该 ID。

查阅位置

• SDK 约定（方法签名、阶段配置、链接架构、列类型、连接绑定、键规则、完整示例）→ references/sdk-conventions.md
• 阶段选择 → recommend_datastage_stages(subutterances=[...])
• 阶段属性名称和可接受的值 → datastage_property_lookup(requests=[{"stage": "..."}])
• 各阶段深入说明 → di-agent-knowledge-engine-datastage 技能 stages/
• 流优化 → di-agent-knowledge-engine-datastage 技能 optimization/overview.md
• 自定义阶段（C/C++、Java） → BuildopStage.md、JavaIntegrationStage.md
• Transformer 阶段表达式 → di-agent-knowledge-engine-datastage 技能资源 stages/TransformerStageFunctions/TransformerStageFunctionsOverview.md

编辑

编辑现有流程： 如果更改仅涉及现有结构中的表达式或阶段属性，那么原位编辑 SDK 是合适的做法——使用 retrieve_datastage_flow_code 获取流程，修改该 SDK，然后通过 update_datastage_flow 重新提交。你是在调整已经验证且连线完成的代码，因此这种做法可靠、无损且成本低。

编辑获取到的代码——绝不要手写片段。 提交给 update_datastage_flow 的内容必须是 retrieve_datastage_flow_code 返回的完整 SDK，并在其中插入你的修改，而不是一个独立代码片段。你引用的每个阶段、连线和架构都必须在同一次提交中先定义为变量（获取到的代码已经定义了它们——derive_seg、veh_in、fp 等，因此请复用这些变量名）。扁平化 SDK 语法无法获取现有元素：flow.links["x"]、flow.stages["y"] 和 link.schema 都会被拒绝，并报错“Unsupported statement type”／“variable does not exist”。要向现有连线添加列，请在获取到的代码中修改该连线现有的 create_schema()／add_field(...) 行；要添加结构，请追加引用已定义变量的 flow.add_stage(...)／stage.connect_output_to(...)／link.create_schema() 语句。如果提交未通过验证，请修正并重新提交整段代码；不要重新获取元素，也不要从头重新编写。

如果更改涉及添加或修改结构——例如新数据源、联接或不同形状——不要在这里从头手写；这是 SDK 最不可靠的路径。请先在 pyflow（di-agent-flow-pyflow）中搭建新结构，然后在这里精确编辑生成的 SDK，添加 pyflow 无法表达的内容。如果核心形状无法通过 pyflow 忠实表达，请先用 pyflow 搭建最接近的形状，再通过 update_datastage_flow 在这里调整。

版本管理和备份由生命周期路由器的 RECOVER 状态（di-agent-flow-lifecycle/references/recover.md）负责。简而言之：update_datastage_flow 会在覆盖流程前自动创建快照——你不需要先调用 duplicate_asset，否则只会留下一个多余的副本。每个流程最多保留 5 个快照（永远不会删除最旧的快照）；默认情况下，list_datastage_flows 不会列出这些快照；删除流程时也会一并删除它们。

防护措施

• 通过 flow_id 获取流程——retrieve_datastage_flow_code 也接受 flow_name，但这种方式会直接采用名称搜索的第一个匹配项，不会检查名称是否唯一，因此名称有歧义时会悄无声息地返回其他流程的代码
• 绝不要猜测阶段类型、属性名称或枚举值——请使用 recommend_datastage_stages／datastage_property_lookup
• 将所有更改打包在一次提交中——连续提交会相互覆盖
• update_datastage_flow 从不创建流程——如果指定的流程不存在，它会报错；请通过 list_datastage_flows 验证名称。要创建流程，请使用 create_datastage_flow（或 pyflow）——具体选择请参阅生命周期路由器
• 更新成功后，将工具结果中返回的 flow_link 作为可点击链接显示，方便用户在界面中打开流程
• 正文中提及阶段属性名称时——使用 datastage_property_lookup 中的 User friendly name（例如“Number of rows (per partition)”），而不是内部标识符（nrecs）；仅在代码块中或用户询问 SDK 属性时显示内部名称
• 参数化流程： update_datastage_flow 会在重建流程前加载现有流程，因此通过 SDK 编辑时会保留参数集注册和本地参数。如果作业在编辑后意外丢失参数绑定，请调用 get_flow_parameter_references 检查已注册的内容，并使用 attach_parameter_set_to_flow／add_local_parameter 恢复缺失的条目。