building-flows

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Building Flows

构建流程

A flow moves data from one or more source systems to one or more destination systems. It runs on a schedule, in response to events (webhooks, listeners), or when triggered by another flow. Flows are the primary way integrations get work done in Celigo.

A flow has page generators (exports that fetch data) and page processors (imports and lookups that process each record). Processors run sequentially in a flat list, or conditionally through routers that branch records to different paths. These processing pipeline mechanics -- routers, branches, page processors, response mapping -- are shared with APIs and tools (see

building-apis

and

building-tools

Flow（流程）负责将数据从一个或多个源系统移动到一个或多个目标系统。它可以按计划运行、响应事件（webhooks、监听器）运行，或者由另一个Flow触发运行。在Celigo中，Flow是实现集成工作的主要方式。

每个Flow包含page generators（页面生成器）（用于获取数据的导出任务）和page processors（页面处理器）（用于处理每条记录的导入与查询任务）。处理器可以按扁平列表顺序运行，也可以通过路由器根据条件将记录分流到不同路径。这些处理管道机制——路由器、分支、页面处理器、响应映射——与API和工具共享（详见

building-apis

和

building-tools

）。

What Starts a Flow

Flow的触发方式

Flows start themselves -- this is the biggest thing that separates them from APIs (invoked by an HTTP caller) and tools (invoked by a consumer). Every flow begins with one or more page generators, of two kinds:

Scheduled exports -- the flow runs on a cron cadence and each run pulls from the source: everything (full sync), only what changed since the last run (delta sync), records matching a query, or files landed in an FTP/SFTP/S3 folder. The right primitive for batch work: nightly reconciliations, hourly delta syncs, backfills, off-peak windows.
Listeners -- the source pushes to the flow. A webhook fires (or a NetSuite/Salesforce native real-time event triggers) and the payload immediately starts flowing. No schedule; the flow runs as events arrive. The right primitive for event-driven work ("when X happens, do Y"), especially when latency matters.

A flow can mix both, and multi-generator designs are common:

Real-time plus batch safety net -- a listener catches events as they fire; a scheduled export reconciles at off-peak hours, catching up after webhook outages
Consolidating sources -- customers from Salesforce AND HubSpot, each with its own generator, feeding the same downstream pipeline
Different slices of the same source -- one export pulls new records, another pulls updated records, when the API exposes them separately

If the requirement is "every night at 2 AM, do X" or "when a webhook arrives, do Y" -- that lives on a flow. APIs and tools have no schedule and no listener; they only run when invoked.

Flow会自行启动——这是它与API（由HTTP调用方触发）和工具（由使用者触发）最大的区别。每个Flow都以一个或两种类型的page generators开头：

定时导出——Flow按cron周期运行，每次运行从源系统拉取数据：全量同步、仅拉取上次运行后变更的数据（增量同步）、匹配查询条件的记录，或是FTP/SFTP/S3文件夹中的文件。适用于批量任务：夜间对账、每小时增量同步、数据回填、非高峰时段处理。
监听器——源系统主动向Flow推送数据。webhook触发（或NetSuite/Salesforce原生实时事件触发）后，负载立即进入Flow处理。无需调度，Flow随事件到达实时运行。适用于事件驱动型工作（“当X发生时，执行Y”），尤其是对延迟敏感的场景。

Flow可以同时使用两种触发方式，多生成器设计十分常见：

实时触发+批量兜底——监听器捕获实时事件；定时导出在非高峰时段进行对账，弥补webhook故障期间的遗漏数据
多源合并——来自Salesforce和HubSpot的客户数据，各自由独立生成器获取，汇入同一下游管道
同一源的不同数据切片——一个导出任务拉取新记录，另一个拉取更新记录，适用于API将两类数据分开暴露的场景

如果需求是“每天凌晨2点执行X”或“当webhook到达时执行Y”，这类场景都适合用Flow实现。API和工具没有调度或监听器，仅在被调用时运行。

Fetched Data Needs a Downstream Consumer

获取的数据需要下游消费者

A common design mistake: ending a flow on an import that fetches data back from a remote system (a preview call, a query, a lookup-shaped POST) and relying on response mapping to capture the result. Response mapping makes fields visible to the NEXT step -- if no next step exists, the captured data is discarded when the run ends and nobody sees it.

When the requirement says "preview / estimate / retrieve / fetch / check / look up", the design needs at least one of:

A write-back import to the source system (most common) -- e.g. source export -> preview import -> update import that writes the captured fields onto the source record
A persistent destination the user named (file to S3/SFTP, email, database)
A router or AI agent step that consumes the captured data within the same run

A two-step export -> fetch-shaped-import flow with nothing after it is a smell -- re-read the intent for where the fetched data should end up. The same applies in reverse: capturing a created record's ID via response mapping is only useful if a later step writes it somewhere.

一个常见的设计误区：Flow以从远程系统获取数据的导入任务结束（预览调用、查询、类查询的POST请求），并依赖响应映射捕获结果。响应映射仅能让字段对下一个步骤可见——如果没有后续步骤，捕获的数据会在运行结束后被丢弃，无人可见。

当需求包含“预览/估算/检索/获取/检查/查询”时，设计需至少满足以下一种情况：

写回源系统的导入任务（最常见）——例如：源系统导出 → 预览导入 → 更新导入，将捕获的字段写回源系统记录
用户指定的持久化目标（如写入S3/SFTP的文件、邮件、数据库）
路由器或AI Agent步骤，在同一次运行中消费捕获的数据

仅包含“导出 → 类查询导入”两步且无后续操作的Flow存在设计缺陷——需重新明确获取数据的最终去向。反之亦然：通过响应映射捕获创建记录的ID，只有在后续步骤将其写入某处时才有意义。

Flow Topologies

Flow拓扑结构

Linear Flows

线性Flow

A flat

pageProcessors[]

list with no routers. One or more page generators feed records through a sequential chain of page processors. Each processor is either an import (

type: "import"

) or a lookup export (

type: "export"

). Records pass through every step in order. Unique to flows -- APIs and tools always use routers.

无路由器的扁平

pageProcessors[]

列表。一个或多个page generators将记录传入按顺序执行的page processors链。每个处理器要么是导入（

type: "import"

），要么是查询导出（

type: "export"

）。记录按顺序通过每一步。这是Flow独有的结构——API和工具始终使用路由器。

Branching Flows

分支Flow

Page generators feed records into

routers[]

instead of

pageProcessors[]

. Each router evaluates records against branch conditions and routes them to matching branches. Branches contain their own

pageProcessors[]

and can chain to other routers via

nextRouterId

Two routing modes (shared with APIs and tools):

Input filters (
```
routeRecordsUsing: "input_filters"
```
) -- S-expression rules on each branch; last branch can omit filter as a catch-all
Script-based (
```
routeRecordsUsing: "script"
```
) -- a JavaScript function returns the branch name

Flows support both

first_matching_branch

and

all_matching_branches

routing. APIs only support

first_matching_branch

. Tools support

first_matching_branch

only.

A flow uses EITHER

pageProcessors

(linear) OR

routers

(branching) at the top level -- not both.

When a branching flow needs linear steps before the branch point (e.g., a lookup enrichment or AI classification that all branches depend on), use a pass-through router: a single-branch router with

nextRouterId

pointing to the branching router. Omit

routeRecordsTo

and

routeRecordsUsing

on the pass-through router -- including them makes it appear as a filter-based branch in the UI. The API defaults are sufficient.

page generators将记录传入

routers[]

而非

pageProcessors[]

。每个路由器根据分支条件评估记录，并将其路由到匹配的分支。分支包含自己的

pageProcessors[]

，并可通过

nextRouterId

链接到其他路由器。

两种路由模式（与API和工具共享）：

输入过滤器（
```
routeRecordsUsing: "input_filters"
```
）——每个分支使用S表达式规则；最后一个分支可省略过滤器作为兜底分支
基于脚本（
```
routeRecordsUsing: "script"
```
）——JavaScript函数返回分支名称

Flow支持

first_matching_branch

（匹配第一个分支）和

all_matching_branches

（匹配所有分支）两种路由模式。API仅支持

first_matching_branch

，工具也仅支持

first_matching_branch

。

一个Flow在顶层只能使用

pageProcessors

（线性）或

routers

（分支）中的一种，不能同时使用。

当分支Flow需要在分支点前执行线性步骤（如所有分支都依赖的查询增强或AI分类）时，可使用直通路由器：单分支路由器，通过

nextRouterId

指向分支路由器。省略直通路由器的

routeRecordsTo

和

routeRecordsUsing

——设置这两个字段会使其在UI中显示为基于过滤器的分支。API默认配置即可满足需求。

Abstract / Instance Flows

抽象/实例Flow

A template/inheritance model. An abstract flow (

isAbstract: true

) defines the complete graph but cannot execute. Instance flows (

_abstractFlowId

) inherit the graph and customize via an

overrides

object (connections, schedules, mappings, filters).

Use when the same flow structure is deployed across multiple regions, tenants, or environments with different connections or parameters.

模板/继承模型。抽象Flow（

isAbstract: true

）定义完整的流程图谱，但无法执行。实例Flow（

_abstractFlowId

）继承图谱，并通过

overrides

对象自定义（连接、调度、映射、过滤器）。

适用于相同流程结构需部署到多个区域、租户或环境，且各环境连接或参数不同的场景。

Quick Reference

快速参考

Flow Type Decision Matrix

Flow类型决策矩阵

Pattern Structure Key fields Read schema

Linear

Flat processor list

Pattern	Structure	Key fields	Read schema
Linear	Flat processor list	`pageGenerators[]` , `pageProcessors[]`	`request.yml` , `page-generator.yml` , `page-processor.yml`
Branching (routers)	Routers with conditional branches	`pageGenerators[]` , `routers[]`	+ `router.yml` , `branch.yml`
Abstract / Instance	Template + per-instance overrides	`isAbstract: true` / `_abstractFlowId` , `overrides`	+ `overrides-helper.yml` , `overrides.yml`

pageGenerators[]

pageProcessors[]

request.yml

page-generator.yml

page-processor.yml

Branching (routers)

Routers with conditional branches

pageGenerators[]

routers[]

router.yml

branch.yml

Abstract / Instance

Template + per-instance overrides

isAbstract: true

_abstractFlowId

overrides

overrides-helper.yml

overrides.yml

模式结构关键字段参考Schema

线性

扁平处理器列表

模式	结构	关键字段	参考Schema
线性	扁平处理器列表	`pageGenerators[]` , `pageProcessors[]`	`request.yml` , `page-generator.yml` , `page-processor.yml`
分支（路由器）	带条件分支的路由器	`pageGenerators[]` , `routers[]`	+ `router.yml` , `branch.yml`
抽象/实例	模板+实例自定义配置	`isAbstract: true` / `_abstractFlowId` , `overrides`	+ `overrides-helper.yml` , `overrides.yml`

pageGenerators[]

pageProcessors[]

request.yml

page-generator.yml

page-processor.yml

分支（路由器）

带条件分支的路由器

pageGenerators[]

routers[]

router.yml

branch.yml

抽象/实例

模板+实例自定义配置

isAbstract: true

_abstractFlowId

overrides

overrides-helper.yml

overrides.yml

Minimum Required Fields

必填字段

Every flow needs at minimum:

```
name
```
-- display name
```
_integrationId
```
-- parent integration
```
disabled: true
```
-- always create disabled
```
pageGenerators[]
```
-- at least one entry with
```
_exportId
```
Either
```
pageProcessors[]
```
(linear) or
```
routers[]
```
(branching) -- never both

每个Flow至少需要以下字段：

```
name
```
——显示名称
```
_integrationId
```
——所属集成
```
disabled: true
```
——创建时默认设为禁用
```
pageGenerators[]
```
——至少包含一个带
```
_exportId
```
的条目
要么
```
pageProcessors[]
```
（线性）要么
```
routers[]
```
（分支）——不可同时使用

Which Schemas to Read

需参考的Schema

Always read:

request.yml -- base flow fields
page-generator.yml -- export sources, per-generator schedules, delta coordination
page-processor.yml -- import/export steps with responseMapping and hooks

Add for branching flows:

router.yml -- routing strategy, record distribution mode
branch.yml -- input filters, per-branch processors, chaining

Add if response mapping is needed:

response-mapping.yml -- extract/generate pairs for carrying data between steps

All available schemas (in references/schemas/):

Base fields (all flows): request.yml
Response shape: response.yml
Page generators: page-generator.yml
Page processors: page-processor.yml
Response mapping: response-mapping.yml
Routers: router.yml
Branches: branch.yml
Abstract flow helpers: overrides-helper.yml
Instance overrides: overrides.yml
Cloning: clone-request.yml, clone-response.yml

必看：

request.yml——Flow基础字段
page-generator.yml——导出源、生成器级调度、增量同步协调
page-processor.yml——带响应映射和钩子的导入/导出步骤

分支Flow额外参考：

router.yml——路由策略、记录分发模式
branch.yml——输入过滤器、分支处理器、链式调用

需要响应映射时参考：

response-mapping.yml——步骤间数据传递的提取/生成规则对

所有可用Schema（位于references/schemas/）：

基础字段（所有Flow）： request.yml
响应结构： response.yml
页面生成器： page-generator.yml
页面处理器： page-processor.yml
响应映射： response-mapping.yml
路由器： router.yml
分支： branch.yml
抽象Flow辅助工具： overrides-helper.yml
实例自定义配置： overrides.yml
克隆： clone-request.yml, clone-response.yml

Related Skills

How to Build a Flow

如何构建Flow

1. Plan the flow

1. 规划Flow

Before creating anything, decide what kind of operation this is:

Decision tree:

Modifying an existing flow's step config (export settings, import mappings, scripts) -- work on the step directly, not the flow. Use
```
celigo exports set
```
,
```
celigo imports set
```
, or the relevant skill (configuring-exports, configuring-imports, writing-scripts, writing-mappings)
Modifying an existing flow's structure (add/remove steps, change schedule, rename) -- GET the flow, modify the structure, PUT it back. Don't rebuild from scratch
Building a new flow where every step is known -- build directly, bottom-up (skip to step 2)
Any ambiguity about what steps are needed -- design first. List every system, every data direction, every step before writing any JSON

Design checklist (when ambiguity exists):

What source systems? What destination systems?
What data moves between them, in which direction?
How often? (cron schedule, webhook trigger, on-demand)
What happens when a step fails? (
```
proceedOnFailure
```
, error notifications)
Do downstream steps need data from upstream responses? (response mapping)
Is this a one-off or a reusable template? (abstract/instance flow)
Sandbox or production? (never mix --
```
sandbox: true
```
flows only use
```
sandbox: true
```
connections)

在创建任何内容前，先确定操作类型：

决策树：

修改现有Flow的步骤配置（导出设置、导入映射、脚本）——直接修改步骤，而非Flow本身。使用
```
celigo exports set
```
、
```
celigo imports set
```
或相关技能（configuring-exports、configuring-imports、writing-scripts、writing-mappings）
修改现有Flow的结构（添加/删除步骤、更改调度、重命名）——先获取Flow信息，修改结构后再更新。无需从头重建
构建所有步骤明确的新Flow——直接自底向上构建（跳至步骤2）
对所需步骤存在疑问——先设计。在编写任何JSON前，列出所有涉及的系统、数据流向、每个步骤

设计检查清单（存在疑问时）：

涉及哪些源系统和目标系统？
哪些数据在系统间流转，方向如何？
运行频率？（cron调度、webhook触发、按需运行）
步骤失败时如何处理？（
```
proceedOnFailure
```
、错误通知）
下游步骤是否需要上游响应的数据？（响应映射）
是一次性任务还是可复用模板？（抽象/实例Flow）
沙箱环境还是生产环境？（切勿混用——
```
sandbox: true
```
的Flow仅使用
```
sandbox: true
```
的连接）

2. Identify the integration

2. 确定集成

Every flow belongs to an integration (the container). Find or create the integration first.

bash

celigo integrations list

每个Flow都属于一个集成（容器）。先找到或创建集成。

bash

celigo integrations list

3. Check for existing patterns

3. 检查现有模式

Before building from scratch, check what already exists in the account and marketplace.

bash

undefined

在从头构建前，先检查账户和市场中已有的资源。

bash

undefined

Search for similar resources in the account index

在账户索引中搜索类似资源

celigo account search "<keyword>"

Show what an existing resource uses and what uses it

查看现有资源的依赖关系

celigo account dependencies flow <id>

Find orphaned resources, offline connections, untriggered flows

查找孤立资源、离线连接、未触发的Flow

celigo account lint

Search marketplace for pre-built integration templates

在市场中搜索预构建的集成模板

celigo templates marketplace

Preview a template before installing

安装前预览模板

celigo templates preview <id> --summary


The account index auto-refreshes when stale (>4 hours). Force a fresh snapshot with `celigo account snapshot`.

celigo templates preview <id> --summary


账户索引会在过期（超过4小时）时自动刷新。可使用`celigo account snapshot`强制生成新快照。

4. Build the connections, exports, and imports

4. 构建连接、导出和导入任务

Flows reference existing resources. Build bottom-up: connections first, then exports and imports that use those connections, then the flow that wires them together.

bash

celigo connections list
celigo exports list
celigo imports list

For every step, match the adaptor to the target application -- raw HTTP is the fallback, not the default. Use the native adaptor when one exists (NetSuite, Salesforce, databases, FTP/S3); otherwise check for a pre-built HTTP connector (550+ apps:

celigo http-connectors list

) and build the connection from it; hand-write HTTP config from public API docs only when no connector exists or it doesn't cover the endpoint. See configuring-exports > Check for a pre-built connector and configuring-imports > Check for a pre-built connector.

See

configuring-exports

and

configuring-imports

for how to build each resource.

Flow引用现有资源。自底向上构建：先创建连接，再创建使用这些连接的导出和导入任务，最后创建将它们关联起来的Flow。

bash

celigo connections list
celigo exports list
celigo imports list

每个步骤都要为目标应用匹配对应的适配器——原生HTTP是 fallback方案，而非默认选项。如果有原生适配器（NetSuite、Salesforce、数据库、FTP/S3）则优先使用；否则检查是否有预构建的HTTP连接器（支持550+应用：

celigo http-connectors list

）并基于它创建连接；仅当没有连接器或连接器不覆盖目标端点时，才根据公开API文档手动编写HTTP配置。详见configuring-exports > Check for a pre-built connector和configuring-imports > Check for a pre-built connector。

构建各资源的方法请参考

configuring-exports

和

configuring-imports

。

5. Choose the topology

5. 选择拓扑结构

Scenario	Topology
All records follow the same path	Linear ( `pageProcessors` )
Records need conditional routing by field values	Branching with input filters
Routing logic requires custom JavaScript	Branching with script router
Records should fan out to all matching paths	Branching with `all_matching_branches`
Same structure across multiple tenants/regions	Abstract + instance flows

Abstract/instance flows: Abstract flows are reusable templates that cannot run directly. Instance flows inherit the abstract's structure and override specific fields (connections, filters, schedules). Use when the same integration pattern repeats across tenants or regions. Create with

isAbstract: true

. Top-level

pageProcessors

are automatically wrapped into a single-branch router. Instance flows reference the abstract via

_abstractFlowId

and specify overrides -- they do NOT use the normal scaffolding process.

_integrationId

is NOT inherited and must be set explicitly on the instance.

场景	拓扑结构
所有记录遵循相同路径	线性（ `pageProcessors` ）
记录需按字段值进行条件路由	带输入过滤器的分支结构
路由逻辑需要自定义JavaScript	带脚本路由器的分支结构
记录需分流到所有匹配路径	带 `all_matching_branches` 的分支结构
相同结构部署到多个租户/区域	抽象+实例Flow

抽象/实例Flow： 抽象Flow是不可直接运行的可复用模板。实例Flow继承抽象Flow的结构，并通过

overrides

字段覆盖特定配置（连接、过滤器、调度）。适用于相同集成模式需部署到多个租户或区域的场景。创建时设置

isAbstract: true

。顶层

pageProcessors

会自动包装为单分支路由器。实例Flow通过

_abstractFlowId

引用抽象Flow，并指定自定义配置——无需使用常规脚手架流程。

_integrationId

不会继承，必须在实例Flow中显式设置。

6. Design the step sequence

6. 设计步骤序列

For each step, decide:

Type --
```
import
```
(write to destination) or
```
export
```
(lookup for enrichment)
Response mapping -- what data from this step's response do downstream steps need? Only add response mapping when downstream steps need fields that aren't already in the record or when field names need to change. If the lookup returns fields with the same names the downstream step expects, skip the response mapping -- it adds complexity without value.
proceedOnFailure -- should the pipeline continue if this step fails?
Hooks -- does this step need a
```
postResponseMap
```
script?

针对每个步骤，确定：

类型——
```
import
```
（写入目标系统）或
```
export
```
（用于增强数据的查询）
响应映射——下游步骤需要此步骤响应中的哪些数据？**仅当下游步骤需要记录中未包含的字段，或需要更改字段名称时，才添加响应映射。**如果查询返回的字段名称与下游步骤预期一致，可跳过响应映射——否则会增加不必要的复杂度。
proceedOnFailure——此步骤失败时，管道是否继续运行？
钩子——此步骤是否需要
```
postResponseMap
```
脚本？

7. Build the flow JSON

7. 构建Flow JSON

Reference the schemas listed in the Quick Reference above for exact field schemas.

参考上述快速参考中的Schema获取准确的字段定义。

8. Configure scheduling

8. 配置调度

Pair

schedule

(6-field cron) with

timezone

(IANA). Omit both for listener/webhook/realtime flows.

timezone

defaults to UTC, which is usually wrong for human-facing schedules ("9 AM every weekday" should survive daylight saving).

Individual page generators can override the flow schedule via their own

schedule

field (e.g. one source syncs hourly, another nightly, in the same flow).

将

schedule

（6字段cron表达式）与

timezone

（IANA时区）配合使用。监听器/webhook/实时Flow可省略这两个字段。

timezone

默认值为UTC，这通常不适用于面向人工的调度（如“每周工作日上午9点”需适应夏令时）。

单个page generators可通过自身的

schedule

字段覆盖Flow的调度（例如，同一Flow中一个源系统每小时同步一次，另一个每天同步一次）。

9. Set the runtime controls

9. 设置运行时控制

Flow-level and per-step switches that change production behavior. APIs and tools have none of these -- they are flow-only.

Control	Where	Default	Flip it when
`proceedOnFailure`	per processor	`false` (a failed record stops there)	The step is non-critical and downstream steps still do meaningful work without it (a Slack notification late in the flow shouldn't block the sync). Keep `false` when downstream depends on this step's output
`skipRetries`	flow, and per generator	`false` (failed jobs retry)	Work is time-sensitive (retrying a stale webhook is meaningless) or non-idempotent (retries risk duplicates). Per-generator override: set it only on the real-time generator
`runPageGeneratorsInParallel`	flow	`false` (generators run sequentially)	Sources are independent and can take the load. Careful: parallel generators hitting the same API can blow rate limits that sequential runs respect
`autoResolveMatchingTraceKeys`	flow	duplicate trace keys raise an error	The source genuinely emits duplicates in normal operation, or the flow is intentionally idempotent. Don't enable it to paper over upstream duplication

Flow级和步骤级的开关，用于调整生产环境行为。API和工具没有这些控制项——仅Flow具备。

控制项	位置	默认值	启用时机
`proceedOnFailure`	每个处理器	`false` （记录失败时停止）	步骤非关键，且下游步骤无需此步骤输出仍能完成有意义的工作（如Flow后期的Slack通知不应阻止同步）。当下游步骤依赖此步骤输出时，保持 `false`
`skipRetries`	Flow级和生成器级	`false` （失败任务会重试）	工作对时间敏感（重试过期的webhook无意义）或操作不具备幂等性（重试可能导致重复数据）。生成器级覆盖：仅在实时生成器上设置
`runPageGeneratorsInParallel`	Flow级	`false` （生成器顺序运行）	源系统相互独立且能承受负载。注意：并行生成器访问同一API可能触发顺序运行不会触发的速率限制
`autoResolveMatchingTraceKeys`	Flow级	重复跟踪键会触发错误	源系统在正常情况下确实会生成重复数据，或Flow具备幂等性。不要为了掩盖上游重复数据而启用此选项

10. Configure chaining (if needed)

10. 配置链式调用（如需）

```
_runNextFlowIds
```
-- trigger other flows when this one completes. The classic use is multi-stage pipelines: "after the customer-master sync finishes, run the orders sync." When a requirement says "X has to happen, then Y", chain two focused flows rather than building one big one
```
_runNextExportIds
```
-- more granular: trigger specific exports inside other flows instead of the whole flow

```
_runNextFlowIds
```
——此Flow完成时触发其他Flow。典型应用是多阶段管道：“客户主数据同步完成后，运行订单同步Flow”。当需求是“必须先执行X，再执行Y”时，将两个专注的Flow链式调用，而非构建一个大型Flow
```
_runNextExportIds
```
——更细粒度：触发其他Flow中的特定导出任务，而非整个Flow

11. Create disabled, verify, enable

11. 禁用状态创建、验证、启用

Always create with

disabled: true

. Verify the structure with

celigo flows get

. Enable only after verification.

创建时始终设置

disabled: true

。使用

celigo flows get

验证结构。仅在验证通过后启用。

CLI Commands

CLI命令

bash

undefined

bash

undefined

CRUD

celigo flows list celigo flows get <id> celigo flows create < flow.json celigo flows update <id> < flow.json celigo flows set <id> key=value [key2=value2 ...] celigo flows delete <id>

Run

celigo flows run <id> [--start-date <ISO8601>] [--end-date <ISO8601>] [--export-ids <ids>] -y

Test run (stage-by-stage)

celigo flows test-run <id> --export <exportId> celigo flows test-run-step-results <id> <runId> <exportOrImportId>

Clone

echo '{"connectionMap":{"oldId":"newId"}}' | celigo flows clone <id> <integrationId> <environmentId> [--flow-group <id>]

Structure manipulation

celigo flows add-generator <id> <exportId> [--schedule '<cron>'] [--index <pos>] celigo flows remove-generator <id> <exportId> celigo flows add-processor <id> <exportOrImportId> [--router <routerId>] [--branch <branchName>] celigo flows remove-processor <id> <exportOrImportId> [--router <routerId>] [--branch <branchName>] celigo flows replace-connection <id> <oldConnectionId> <newConnectionId>

Error management

celigo flows errors <id> <exportOrImportId> celigo flows resolved-errors <id> <exportOrImportId> celigo flows resolve-errors <id> <exportOrImportId> [errorIds] [-y] celigo flows retry-errors <id> <exportOrImportId> [retryDataKeys] [-y] celigo flows assign-errors <id> <exportOrImportId> <email> [errorIds] [-y] celigo flows delete-resolved-errors <id> <exportOrImportId> [errorIds] [-y] celigo flows error <id> <exportOrImportId> <errorId> [--retry-data] [--request-detail] celigo flows update-error-data <id> <exportOrImportId> <errorId> celigo flows tag-errors <id> <exportOrImportId> celigo flows error-summary <id> celigo flows error-analysis <id> <exportOrImportId> [--limit <n>]

Debug

celigo flows debug-requests <id> <exportOrImportId> [--since <minutes>] celigo flows debug-request-detail <id> <exportOrImportId> <key> celigo flows enable-execution-logs <id> [--duration <minutes>] celigo flows disable-execution-logs <id> celigo flows execution-logs <id> <jobId> celigo flows query-execution-logs <id> <jobId> --export-or-import-id <id> --group-id <gid> --record-id <rid> celigo flows execution-log-detail <id> <jobId> --export-or-import-id <id> --stage <stage> --group-id <gid> --record-id <rid>

Metadata

celigo flows last-export-date <id>

Integration-level flow management

celigo integrations flow-groups <integrationId> celigo integrations create-flow-group <integrationId> <name> celigo flows set-group <flowGroupingId> <flowIds...>


<!-- TIER:3 -->

celigo integrations flow-groups <integrationId> celigo integrations create-flow-group <integrationId> <name> celigo flows set-group <flowGroupingId> <flowIds...>


<!-- TIER:3 -->

Pre-Submit Checklist

提交前检查清单

Before creating or updating a flow, verify:

```
_integrationId
```
references a real integration (confirm with
```
celigo integrations get <id>
```
)
```
disabled: true
```
is set for initial creation -- an enabled flow with a schedule runs immediately
```
schedule
```
is 6-field cron with seconds:
```
"? */5 * * * *"
```
(first field is always
```
?
```
)
```
pageProcessors[]
```
and
```
routers[]
```
are mutually exclusive -- a flow uses one or the other, never both
Router IDs are unique within the flow (random alphanumeric, e.g.,
```
"N8Q9NX24Sj5"
```
)
Every branch
```
nextRouterId
```
references an existing router
```
id
```
in the same flow

创建或更新Flow前，验证以下内容：

_integrationId

引用真实的集成（使用

celigo integrations get <id>

确认）

初始创建时设置
```
disabled: true
```
——启用且带调度的Flow会立即运行
```
schedule
```
为6字段带秒数的cron表达式：
```
"? */5 * * * *"
```
（第一个字段始终为
```
?
```
）
```
pageProcessors[]
```
和
```
routers[]
```
互斥——Flow只能使用其中一种，不可同时使用
路由器ID在Flow内唯一（随机字母数字，如
```
"N8Q9NX24Sj5"
```
）
每个分支的
```
nextRouterId
```
引用同一Flow中存在的路由器
```
id
```

Gotchas

注意事项

PUT erases omitted fields. Always GET first, modify, then PUT. The
```
set
```
command handles this automatically.
pageProcessors
and
routers
are mutually exclusive. A flow uses one or the other at the top level. Setting both causes validation errors.
Router IDs must be unique within a flow. Use random alphanumeric strings (e.g.,
```
"N8Q9NX24Sj5"
```
).
```
nextRouterId
```
must reference an existing router
```
id
```
in the same flow.
Create flows with
disabled: true
. An enabled flow with a schedule will run immediately. Enable only after verification.
Build order matters. Connection -> Export -> Import -> Flow. The API rejects references to non-existent resources.
Schedule is 6-field cron with seconds. Format:
```
"? minute hour dayOfMonth month dayOfWeek"
```
. The first field is always
```
?
```
. Common mistake: using 5-field cron without the seconds position.
Instance flows cannot define structure. Do not set
```
pageGenerators
```
,
```
pageProcessors
```
, or
```
routers
```
on instance flows -- these are inherited from the abstract flow. All customizations go through
```
overrides
```
.
Instance flow
overrides
is full-replace on PUT. Omitting an override entry removes it. Always GET, merge changes, then PUT.
Empty
pageProcessors: []
in a branch is the discard pattern. Records matching that branch are dropped. A branch with no
```
inputFilter
```
serves as a catch-all.
responseMapping
uses Transformation 1.0 syntax (extract/generate pairs), not expression-based transforms. Lookup export responses use
```
data[0].fieldName
```
; import responses use
```
_json.fieldName
```
.
Don't add unnecessary transforms on lookups. If the lookup returns fields with the same names the downstream import expects, skip the transform -- the data flows through as-is. Only add a transform when you need to rename fields, reshape nested structures, or drop fields. An identity transform (e.g.,
```
errorId
```
->
```
$.errorId
```
) adds complexity for no benefit.

PUT请求会删除未指定的字段。始终先GET，修改后再PUT。
```
set
```
命令会自动处理此问题。
pageProcessors
和
routers
互斥。Flow在顶层只能使用其中一种，同时设置会导致验证错误。
路由器ID在Flow内必须唯一。使用随机字母数字字符串（如
```
"N8Q9NX24Sj5"
```
）。
```
nextRouterId
```
必须引用同一Flow中存在的路由器
```
id
```
。
创建Flow时设置
disabled: true
。启用且带调度的Flow会立即运行。仅在验证通过后启用。
构建顺序很重要。连接 → 导出 → 导入 → Flow。API会拒绝引用不存在资源的请求。
调度为6字段带秒数的cron表达式。格式：
```
"? minute hour dayOfMonth month dayOfWeek"
```
。第一个字段始终为
```
?
```
。常见错误：使用不带秒数的5字段cron表达式。
实例Flow不可定义结构。不要在实例Flow中设置
```
pageGenerators
```
、
```
pageProcessors
```
或
```
routers
```
——这些继承自抽象Flow。所有自定义配置通过
```
overrides
```
实现。
实例Flow的
overrides
在PUT时会完全替换。省略某个自定义配置条目会将其移除。始终先GET，合并更改后再PUT。
分支中
pageProcessors: []
为空是丢弃模式。匹配该分支的记录会被丢弃。无
```
inputFilter
```
的分支作为兜底分支。
responseMapping
使用Transformation 1.0语法（提取/生成规则对），而非基于表达式的转换。查询导出响应使用
```
data[0].fieldName
```
；导入响应使用
```
_json.fieldName
```
。
不要在查询中添加不必要的转换。如果查询返回的字段名称与下游导入任务预期一致，跳过转换——数据会自动流转。仅当需要重命名字段、重构嵌套结构或删除字段时才添加转换。恒等转换（如
```
errorId
```
->
```
$.errorId
```
）会增加不必要的复杂度。

Common Errors

常见错误

Error	Cause	Fix
`"pageProcessors" is not allowed when "routers" is present`	Both `pageProcessors[]` and `routers[]` set on the same flow	Remove one -- use `pageProcessors` for linear, `routers` for branching
`Invalid reference: _integrationId`	Integration ID does not exist or is misspelled	Verify with `celigo integrations get <id>`
`Invalid reference: _exportId` / `_importId`	Export or import referenced in a page generator/processor does not exist	Create the export/import first, then reference it
`Invalid reference: _connectionId`	Connection ID on an export or import does not exist	Verify with `celigo connections get <id>`
`Duplicate router id`	Two routers in the same flow share the same `id`	Assign unique alphanumeric IDs to each router
`Invalid nextRouterId`	A branch references a router `id` that does not exist in the flow	Ensure `nextRouterId` matches an actual router `id` in the same flow
`Invalid cron expression`	Schedule uses 5-field cron or wrong format	Use 6-field format: `"? /5 * * *"` (seconds field first, always `?` )
Flow runs immediately after creation	Created with `disabled: false` or `disabled` omitted (defaults to enabled)	Always set `disabled: true` on create; enable after verification

错误	原因	修复方案
`"pageProcessors" is not allowed when "routers" is present`	同一Flow中同时设置了 `pageProcessors[]` 和 `routers[]`	删除其中一个——线性Flow用 `pageProcessors` ，分支Flow用 `routers`
`Invalid reference: _integrationId`	集成ID不存在或拼写错误	使用 `celigo integrations get <id>` 验证
`Invalid reference: _exportId` / `_importId`	page generator/processor中引用的导出或导入任务不存在	先创建导出/导入任务，再引用
`Invalid reference: _connectionId`	导出或导入任务中的连接ID不存在	使用 `celigo connections get <id>` 验证
`Duplicate router id`	同一Flow中两个路由器使用相同的 `id`	为每个路由器分配唯一的字母数字ID
`Invalid nextRouterId`	分支引用的路由器 `id` 在Flow中不存在	确保 `nextRouterId` 与同一Flow中存在的路由器 `id` 匹配
`Invalid cron expression`	调度使用5字段cron或格式错误	使用6字段格式： `"? /5 * * *"` （第一个字段为秒数，始终为 `?` ）
Flow创建后立即运行	创建时设置 `disabled: false` 或省略 `disabled` （默认启用）	创建时始终设置 `disabled: true` ；验证通过后再启用

building-flows

Original

Translation

Building Flows

构建流程

What Starts a Flow

Flow的触发方式

Fetched Data Needs a Downstream Consumer

获取的数据需要下游消费者

Flow Topologies

Flow拓扑结构

Linear Flows

线性Flow

Branching Flows

分支Flow

Abstract / Instance Flows

抽象/实例Flow

Quick Reference

快速参考

Flow Type Decision Matrix

Flow类型决策矩阵

Minimum Required Fields

必填字段

Which Schemas to Read

需参考的Schema

Related Skills

相关技能

How to Build a Flow

如何构建Flow

1. Plan the flow

1. 规划Flow

2. Identify the integration

2. 确定集成

3. Check for existing patterns

3. 检查现有模式

Search for similar resources in the account index

在账户索引中搜索类似资源

Show what an existing resource uses and what uses it

查看现有资源的依赖关系

Find orphaned resources, offline connections, untriggered flows

查找孤立资源、离线连接、未触发的Flow

Search marketplace for pre-built integration templates

在市场中搜索预构建的集成模板

Preview a template before installing

安装前预览模板

4. Build the connections, exports, and imports

4. 构建连接、导出和导入任务

5. Choose the topology

5. 选择拓扑结构

6. Design the step sequence

6. 设计步骤序列

7. Build the flow JSON

7. 构建Flow JSON

8. Configure scheduling

8. 配置调度

9. Set the runtime controls

9. 设置运行时控制

10. Configure chaining (if needed)

10. 配置链式调用（如需）

11. Create disabled, verify, enable

11. 禁用状态创建、验证、启用

CLI Commands

CLI命令

CRUD

CRUD

Run

Run

Test run (stage-by-stage)

Test run (stage-by-stage)

Clone

Clone

Structure manipulation

Structure manipulation

Error management

Error management

Debug

Debug

Metadata

Metadata

Integration-level flow management