用Git管理Openflow流程组：从画布到Dev/Prod的版本控制

如果你用过任何 GUI 驱动的管道工具，你很可能经历过类似的情况：你在可视化界面中构建管道，当需要将其迁移到另一个环境时，你将其导出为文件，交给别人，并希望它能干净地导入到另一边。也许你会把它重命名为类似 pipeline_final_v3_actually_final.xml 的文件，然后放到共享文件夹中，希望别人在你之前不要打开它。

Openflow 是 Snowflake 基于 Apache NiFi 构建的托管服务，其中管道被称为 Process Group，你会在本文中看到这个术语。但这里的基本问题并非 NiFi 或 Openflow 所特有，而是导出-再导入这一模式本身。

这种模式有三个问题，无论你从哪个工具导出，它们都不会真正消失。导出的文件不包含运行时状态，所以任何有状态的东西在导入时都会回到初始状态。没有差异比较功能。你不能通过查看两个导出文件来了解它们之间实际发生了什么变化，所以审核意味着盯着画布，希望你能注意到某个被修改的属性。而且整个过程都是手动的。有人导出、有人重命名、有人导入、有人提升。每一步都可能出错，导致开发和生产环境之间的变更出现失误。

Openflow 并没有通过提供更好的导出按钮来解决这个问题。它通过给进程组与 Git 建立实际的关系来解决：分支、提交、差异、Pull Request，等等。本月的内容是关于如何端到端地设置这一切，从裸的 GitHub 仓库到真正的开发到生产环境的提升，参数也会被妥善处理。

我们正在构建的内容

• 将基于 GitHub 的注册表客户端连接到 Controller Settings，使进程组能够针对真实仓库开始版本控制。
• 一个虽小但真实的转换流程，在画布上进行版本控制，并查看实际以 JSON 形式落入 Git 的内容。
• 一个功能分支工作流：在画布上更改、提交到分支、打开 Pull Request、审查差异、合并。
• 一个提升路径，另一个生产运行时从 main 拉取合并后的流程。
• 父级和子级参数上下文，它们共享公共值，但按环境覆盖。

思维转变：进程组是变更单元，而非导出文件

基于模板的 NiFi 直觉是考虑移动文件。导出这个、导入那个，并在安全的地方保留一个版本文件夹。

使用基于 Git 的进程组，进程组本身是版本控制的对象。它在画布上有一个工作状态，在 Git 中有一个已提交状态，就像工作目录与 Git 仓库的关系一样。当你想获得快照时，不需要导出，而是提交。你不需要手动比较两个导出文件，而是打开一个 Pull Request 并阅读差异。变更单元是进程组的配置，而不是你从中生成的工件。

这在提升时最为重要。将变更移到生产环境不是“将此文件导入另一个环境”，而是“将生产进程组指向更新的提交”，这是一个更小、更安全的操作。

架构概览

开发工作在注册表客户端的 GitHub 仓库内的功能分支上进行。一个 Pull Request 将该分支合并到 main。一个生产进程组，针对同一个仓库进行版本控制，当你准备好提升时，更新以从 main 拉取最新的提交。

开发画布和生产画布是两个不同的进程组，指向同一注册表中的同一个存储桶。两者都不拥有流程定义，Git 拥有它。它们之间的区别在于当前同步到的提交，以及在运行时解析环境特定值的参数上下文。

前提条件

1. 一个作为流程存储的 GitHub 仓库，以及一个具有 repo 权限的个人访问令牌（PAT），用于 Openflow 对其进行身份验证。
2. 具有等同于 ACCOUNTADMIN 权限的 Openflow 访问权限，以访问 Controller Settings，从而添加注册表客户端。
3. 一个开发 Openflow 运行时和一个生产 Openflow 运行时（或你视为这两种环境的两个环境），它们都能访问 GitHub。

如果你的组织使用 Azure DevOps 或 GitLab 而不是 GitHub，Openflow 也提供一个 Azure DevOps/GitLab 注册表客户端，其工作方式相同，包括仓库和 PAT。本文中除注册表客户端设置之外的所有内容都同样适用，无论你选择哪个。

设置

创建 GitHub 仓库和 PAT

创建一个空仓库来存放流程定义。它不需要 README 或 .gitignore，Openflow 会填充它。生成一个作用域为该仓库 repo 访问的 PAT，并将其存储在秘密管理器或 Openflow 可以安全读取的地方。不要直接将其粘贴到处理器属性中。

添加 GitHub 注册表客户端

在 Controller Settings 中，转至 Registry Clients 并添加新客户端。

这里的 Repository Branch 是 Openflow 在首次浏览注册表时读取的默认分支。它不是锁。一旦你开始进行更改，各个进程组可以并且将会针对功能分支进行版本控制。

在进程组上启动版本控制

我构建了一个虽小但真实的转换流程，而不是 GenerateFlowFile 玩具，因为两个处理器的循环并不能告诉你关于实际落入 Git 的内容的任何有用信息。以下是重现方法：

1. 在开发画布上创建一个新的进程组，命名为 json-transform-router。
2. 拖入一个 GenerateFlowFile 处理器作为源。设置 Custom Text 为一个小的订单负载，{ "id": "ORD-1042", "email": "[email protected]" }，并通过处理器上的 Insert Header / 动态属性字段向其中添加 record.type 属性，以便下游路由有可依据的条件。
3. 拖入一个 JoltTransformJSON 处理器。设置 Jolt Transform 为 chain 和 Jolt Specification 为下面的 shift 规范，它会将两个字段重命名为下游系统期望的形状。
4. 拖入一个 RouteOnAttribute 处理器。添加一个动态属性 record.type，值为 ${record.type:equals('order')}，这样携带订单记录的 FlowFile 会路由到与其他文件不同的关系。
5. 将处理器连接在一起：GenerateFlowFile 的 success 关系连接到 JoltTransformJSON，其 success 关系再连接到 RouteOnAttribute。终止或路由 RouteOnAttribute 上未匹配的关系，使画布上没有悬空的关系。

三个处理器，一个 JOLT 规范，一个路由决策。小到可以在截图中阅读，真实到本文后面的差异对比有意义。

[
  {
    "operation": "shift",
    "spec": {
      "id": "order_id",
      "email": "customer_email"
    }
  }
]

一个形如 { "id": "ORD-1042", "email": "[email protected]" } 的 FlowFile 从另一端输出时变为 { "order_id": "ORD-1042", "customer_email": "[email protected]" }。没有嵌套，没有花哨，仅重命名就足以在后续查看差异时证明转换已生效。

在开发画布上构建好流程组后，右键点击并选择版本 → 开始版本控制。

提交后，流程组的画布头部会显示版本徽章，而不是通常的“未版本化”状态。这是大多数教程跳过

的部分。以下是实际进入仓库的内容：

{
  "flowContents": {
    "processGroups": [],
    "processors": [
      {
        "identifier": "5e2c...",
        "name": "JoltTransformJSON",
        "type": "org.apache.nifi.processors.jolt.JoltTransformJSON",
        "properties": {
          "Jolt Transform": "chain",
          "Jolt Specification": "[ { \"operation\": \"shift\", \"spec\": { \"id\": \"order_id\", \"email\": \"customer_email\" } } ]"
        },
        "position": { "x": 320.0, "y": 128.0 }
      },
      {
        "identifier": "8a91...",
        "name": "RouteOnAttribute",
        "type": "org.apache.nifi.processors.standard.RouteOnAttribute",
        "properties": {
          "record.type": "${record.type:equals('order')}"
        }
      }
    ],
    "connections": [ /* processor-to-processor wiring, by identifier */ ]
  },
  "parameterContexts": {},
  "flowEncodingVersion": "1.0"
}

处理器属性、画布位置和连接线路都会序列化到该 JSON 中。处理器标识符在注册表内是稳定的，这使得差异对比有意义。更改 JOLT 规范，差异会精确显示该字符串的变化，而不是每个处理器像素级移动的噪音墙。

开发 → 审查 → 生产循环

在画布上更改的功能分支

对于下一个更改，我不需要先动 GitHub。右键点击版本化的流程组并选择版本 → 创建分支，这会弹出一个对话框，显示当前分支（main）和新分支名称字段 feature/add-error-routing。Openflow 会在 GitHub 中创建分支，并一步将流程组切换为基于该分支进行版本控制。之后每次从画布提交都会落到功能分支，而不是直接提交到 main。

如果你在 GitHub 中开启了分支保护，Openflow 不会允许你直接提交到受保护的分支，这对本仓库的 main 分支是值得启用的，就像对应用代码一样。

分支创建后，我在画布上进行了实际更改：一个 LogAttribute 处理器连接到 RouteOnAttribute 的未匹配关系，这样未通过 record.type 检查的行会被记录，而不是静默地从画布上消失。流程组的状态栏立即捕捉到更改，显示尚未进入 Git 的本地修改计数。

要提交，再次右键点击流程组并选择版本 → 提交本地更改。这会打开一个对话框来编写提交消息，与初始版本控制提交相同，并将结果直接推送到 feature/add-error-routing。

打开 PR 并审查差异

分支推送后，像往常一样在 GitHub 中打开拉取请求。这就是模板导出中“无差异”问题真正解决的地方，但需要先进行一项设置：将 Snowflake 的 Flow Diff GitHub Action 放入流存储仓库的 .github/workflows/flowdiff.yml 中。一旦该工作流存在，针对该仓库的每个 PR 都会在 PR 对话中直接渲染出可视化的、人类可读的流程差异，新增的处理器、更改的属性以及重新连接的线路都会以真实图表而非原始 JSON 块显示。

从未打开过画布的审查者可以查看该差异，并准确理解更改了什么：一个新的 RouteOnAttribute 关系，一个连接到它的新 LogAttribute 处理器，现有转换逻辑没有变化。这就是模板导出从未给你的审查体验。

合并到 main

批准、合并、完成。main 现在已包含失败路由更改的提交。

提升到生产环境

生产环境有自己的流程组，版本化到相同的存储桶和流程名称，当前同步到失败路由更改之前的提交。提升不是重新部署。它是告诉该流程组更改其版本。

右键点击生产流程组并选择版本 → 更改版本，然后选择 main 上的最新提交。

Openflow 会将当前运行配置与目标提交进行差异对比，并应用差异。它不会拆毁并重建整个流程组。只有画布上更改的内容才会在生产环境中更改。

跨环境的参数上下文

流程本身在开发和生产之间应完全相同。不同之处在于配置：要连接的数据库、要读取的存储桶、重试次数等。这正是参数上下文的用途，Openflow 支持它们之间的继承。

每个参数上下文都与一个流程组一一对应，这就是为什么继承模型在这里很重要。你不是在所有流程之间共享一个巨大的上下文。你为每个流程组提供自己的上下文，并让父子关系传递那些在各环境中应保持相同的值。

我设置了一个父上下文 shared-pipeline-params，包含各处相同的值：重试次数、批次大小、时区设置。每个环境都有自己的子上下文，继承父上下文并仅覆盖实际不同的部分。

开发环境有自己的子上下文 dev-params，继承相同的 retry.count 和 batch.size，将 target.database 覆盖为 DEV_DB，并将 target.warehouse 覆盖为较小的仓库。流程组引用适合其环境的子上下文。无需手动保持共享值同步，因为它们只存在于一处。

限制与注意事项

• 流程 ID 不能跨注册表移植。 针对一个注册表客户端进行版本控制的流程组不能简单地重新指向另一个注册表并期望历史记录能够延续。如果你要迁移注册表，请计划重新开始版本控制，并将旧注册表中的历史视为存档，而不是可以拖动前进的东西。
• 回滚不会恢复处理器状态。将版本更改为较旧的提交会恢复配置，但不会恢复运行中的状态，例如有状态处理器的最后水位线或游标。在回滚之前，要了解处理器持有什么状态，并准备好重置或重新播种。
• 参数更改不会像流程结构那样进行版本控制。最近的发布说明明确指出了这一点：编辑参数上下文的值本身不会生成流程提交，只有流程组的结构更改才会生成。人们很容易认为参数编辑会像处理器更改一样被跟踪。事实并非如此。将参数更改视为独立的变更管理事项，而不是由基于 Git 的版本控制自动涵盖的。
• 分支保护是您的责任，不是 Openflow 的责任。如果仓库未强制执行保护规则，没有什么能阻止某人直接对主分支进行流程组的版本控制。像对待其他重要仓库一样设置保护规则。

总结

导出为模板，然后重新导入的工作流程已完全消失。取而代之的是流程组与 Git 之间的真实关系：用于进行中工作的分支、审阅者真正能阅读的差异、任何内容进入生产环境之前的 PR，以及作为版本更改而非文件移动的升级步骤。参数上下文和 Secrets Manager 处理了仅靠版本控制无法处理的部分，使得完全相同的流程定义可以在两个环境中正确运行，而无需在 JSON 中携带任何一个秘密。

端到端设置的检查清单：

• [ ] 为流程存储创建一个 GitHub 仓库，并为其创建具有相应权限的 PAT
• [ ] 在控制器设置中添加 GitHub（或 Azure DevOps）注册表客户端
• [ ] 在流程组上启动版本控制，确认 JSON 按预期落入仓库
• [ ] 创建分支，在画布上编辑，提交到分支，打开 PR，使用 Flow Diff 审查，合并
• [ ] 通过“更改版本”将生产流程组指向已合并的提交
• [ ] 为共享值设置父参数上下文，并为每个环境设置子上下文
• [ ] 在流程存储仓库中对主分支启用分支保护

参考： Openflow 中自定义流程的版本控制 — 本文涵盖的官方 Snowflake 文档，涉及注册表客户端、Flow Diff GitHub Action 和参数上下文继承。

后续步骤：

• 通过 Secrets Manager（AWS Secrets Manager、Azure Key Vault 或类似服务）连接敏感参数，而不是将它们作为字面值存储在参数上下文中，这样凭据就不会触及提交到 Git 的 JSON
• 现在流程结构和环境配置都能干净地提升，看看 CI 检查在人员打开差异之前可以对 PR 进行哪些实际验证

这是关于使用 Snowflake Openflow 构建数据管道的系列文章的一部分。

后续步骤

• 了解更多信息，请访问 docs.snowflake.com
• 注册 Snowflake 试用

如果觉得有用，请在 LinkedIn 上关注我，获取更多数据工程和 Snowflake AI 数据云用例。

《像代码一样对待 Openflow：用于流程组的基于 Git 的工作流》 最初发布在 Snowflake Builders Blog：数据工程师、应用开发者、AI 与数据科学 在 Medium 上，人们在那里通过点赞和回复继续讨论这个故事。