magic-etl

Name: magic-etl
Author: stahura

github.com/stahura数据与数据库24178 次安装26 周前更新

代码仓库

关于 magic-etl

magic-etl 介绍如何通过 API 和 Java CLI 创建、更新及运行 Domo Magic ETL 数据流。它面向以 JSON DAG 定义转换，并需管理执行与画布布局的 Domo 用户。

主要功能

01通过 REST API 创建和更新数据流
02用 Java CLI 列出、导出、运行和检查数据流
03以 JSON DAG 定义数据流操作
04配置数据集输入、输出和执行行为
05用分区和定位的磁贴组织可视画布

适用场景

01基于数据集输入和转换构建 Magic ETL 数据流
02导出现有数据流定义以检查其结构
03运行数据流并查看执行状态和结果

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/stahura/domo-ai-vibe-rules --skill magic-etl

安装后在 Claude Code 中输入/magic-etl确认已加载。

详细介绍

在 Domo 中以编程方式创建 Magic ETL 数据流

概述

Domo 中的 Magic ETL 数据流可以完全通过 API 进行创建、更新和执行。数据流被定义为一个 JSON 文档，描述了一个由操作组成的有向无环图（DAG）——输入节点、转换和输出节点。

CLI 优先：读取/运行操作使用 Java CLI，创建操作回退到 REST API。

操作	使用 CLI？	使用 REST API？
列出数据流	`list-dataflow -dt MAGIC`	—
导出数据流定义	`list-dataflow -i <ID> -d -f <FILE>`	—
运行数据流	`dataflow-run-now -i <ID>`	`POST .../executions`（对新数据流更可靠）
检查执行状态	`list-dataflow -i <ID> -e -l 1`	`GET .../executions?limit=1`
创建数据流	—	`POST /api/dataprocessing/v1/dataflows`（开发者令牌可用，立即执行）
更新数据流	`set-dataflow-properties -i <id> -d <file>`	`PUT /api/dataprocessing/v1/dataflows/<ID>`
重命名/启用	`set-dataflow-properties -i <id> -n/-e/-s`	`PUT /api/dataprocessing/v1/dataflows/<ID>/patch`

重要： 通过 API（POST）创建数据流可使用开发者令牌（2026 年 4 月已确认）。创建的数据流可立即执行——无需在 UI 中保存。domo（ryuu）CLI 仅用于 Custom App 发布；数据流操作请使用 Java CLI（domoutil.jar）。

数据流的 CLI 命令

列出所有 Magic ETL 数据流

list-dataflow -dt MAGIC

获取数据流定义（导出为 JSON）

list-dataflow -i <DATAFLOW_ID> -d -f <OUTPUT_FILE>

这会导出完整的 JSON 定义，包括所有操作、输入、输出和 GUI 位置。这是理解 JSON 结构的最佳方式——导出一个现有数据流并研究它。

运行数据流

dataflow-run-now -i <DATAFLOW_ID>

列出数据流执行记录

list-dataflow -i <DATAFLOW_ID> -e -l <LIMIT>

通过 API 检查执行状态

curl -s "https://<instance>.domo.com/api/dataprocessing/v1/dataflows/<ID>/executions?limit=1&offset=0" \
  -H "x-domo-developer-token: <TOKEN>"

返回执行状态（SUCCESS、FAILED_DATA_FLOW、CREATED、RUNNING）、行数、错误等。

通过 API 创建数据流

端点

POST https://<instance>.domo.com/api/dataprocessing/v1/dataflows

认证（2026 年 3 月）： 此端点需要 基于 SID 的认证，而非开发者令牌。开发者令牌会得到 403 Forbidden。请使用 SID 交换：刷新令牌 → oauth2/token → 访问令牌 → oauth2/sid → SID。

请求头：

Content-Type: application/json
X-Domo-Authentication: <SID>

请求体必须包含 "databaseType": "MAGIC"。创建的数据流可立即执行——详情见注意事项 #7 和 #11。

注意： x-domo-developer-token 适用于所有数据流操作，包括 POST（创建）、GET、PUT 和执行（2026 年 4 月已确认）。

更新现有数据流

PUT https://<instance>.domo.com/api/dataprocessing/v1/dataflows/<DATAFLOW_ID>

请求头和请求体格式与 POST 相同。需包含完整定义。

数据流 JSON 结构

顶层字段

{
  "name": "My Magic ETL",
  "databaseType": "MAGIC",
  "responsibleUserId": 149955692,
  "draft": false,
  "enabled": true,
  "runState": "ENABLED",
  "engineProperties": {
    "kettle.mode": "STRICT"
  },
  "inputs": [ ... ],
  "outputs": [ ... ],
  "actions": [ ... ]
}

字段	描述
`name`	数据流的显示名称
`databaseType`	对于 Magic ETL 必须为 `"MAGIC"`
`responsibleUserId`	所有者的数字用户 ID
`draft`	已发布的数据流为 `false`
`enabled`	为 `true` 时允许执行
`runState`	为 `"ENABLED"` 时允许计划运行
`engineProperties`	`{"kettle.mode": "STRICT"}` 为标准配置
`inputs`	输入数据集引用的数组
`outputs`	输出数据集引用的数组
`actions`	操作节点（DAG）的数组

顶层 `gui` 字段（画布布局与分区）

gui 字段控制可视化画布布局，包括将相关磁贴分组的彩色 Section 区域。当 useGraphUI: true 时，画布的 elements 数组控制磁贴定位（覆盖操作级别的 gui.x/gui.y）。

{
  "gui": {
    "version": "1.0",
    "canvases": {
      "default": {
        "canvasSettings": {
          "coarserGrid": false,
          "hideCoarserGridPopUp": false,
          "backgroundVariant": "None"
        },
        "elements": [
          { "type": "Section", ... },
          { "type": "Tile", ... }
        ],
        "disabledActions": []
      }
    },
    "useGraphUI": true
  }
}

Section 元素（彩色区域背景）

Section 是彩色矩形区域，用于在画布上对相关 tile 进行视觉分组：

{
  "id": "a-unique-uuid",
  "type": "Section",
  "x": 56,
  "y": 72,
  "width": 1320,
  "height": 288,
  "name": "Work Orders Denormalization",
  "backgroundColor": "var(--colorChartBlue6)"
}

字段	描述
`id`	Section 的唯一 UUID
`type`	必须为 `"Section"`
`x`、`y`	画布上的绝对位置（左上角）
`width`、`height`	彩色区域的尺寸（像素）
`name`	显示在 Section 标题中的标签
`backgroundColor`	区域颜色的 CSS 变量（见下表）

可用的 Section 背景颜色

CSS 变量	颜色	建议用途
`var(--colorChartBlue6)`	浅蓝色	输入/暂存流水线
`var(--colorChartGreen6)`	浅绿色	质量/校验流水线
`var(--colorChartPurple6)`	浅紫色	输出/发布流水线
`var(--colorChartOrange6)`	浅橙色	转换/增强流水线
`var(--colorChartRed6)`	浅红色	过滤/排除流水线
`var(--colorChartYellow6)`	浅黄色	共享/维度表

后缀 6 表示最浅的色调——非常适合用作 Section 背景，以便 tile 标签保持可读。

Tile 元素（操作位置）

每个 action 节点在画布中都有一个对应的 Tile 元素。Tile 可以 被设为 Section 的子元素，此时其 x/y 是 相对于 Section 的位置：

{
  "id": "LoadFromVault-work_orders",
  "type": "Tile",
  "x": 72,
  "y": 56,
  "parentId": "section-uuid-here",
  "color": null,
  "colorSource": null
}

字段	描述
`id`	必须与 action 的 `id` 字段匹配
`type`	必须为 `"Tile"`
`x`、`y`	位置——如果没有 `parentId` 则为绝对坐标，如果设置了 `parentId` 则相对于父 Section
`parentId`	父 Section 元素的 UUID（无父级的 tile 可省略）
`color`	tile 图标可选的颜色整数代码

重新设置父级的公式： 将 tile 移入 section 时，需要把绝对坐标转换为相对坐标：

relativeX = absoluteX - section.x
relativeY = absoluteY - section.y

没有 parentId 的 Tile 会按画布绝对坐标渲染，并悬浮在任何 section 之外。

创建 Dataflow 时务必添加 Section

当构建具有多个逻辑分支或处理阶段的数据流时，务必添加 Section 元素 来组织视觉布局。这是一项最佳实践，可以让数据流一目了然。按以下方式对 tile 进行分组：

流水线分支（例如，每个事实表的 join 链拥有自己的 section）
处理阶段（例如，"Input/Staging"、"Transforms"、"Output"）
共享资源（例如，跨分支使用的维度表）

为每个 section 分配不同的颜色，以便在视觉上加以区分。

Inputs 数组

每个 input 都引用一个为数据流提供数据的 Domo 数据集：

"inputs": [
  {
    "dataSourceId": "422efbf4-6c96-4576-907b-eacae8379d5d",
    "executeFlowWhenUpdated": false,
    "dataSourceName": "RAW | SFDC | Accounts",
    "onlyLoadNewVersions": false,
    "recentVersionCutoffMs": 0
  }
]

字段	描述
`dataSourceId`	输入数据集的 UUID
`dataSourceName`	显示名称（便于阅读）
`executeFlowWhenUpdated`	设为 `true` 时，当此输入更新时自动触发数据流
`onlyLoadNewVersions`	设为 `true` 时，仅处理新的数据版本

Outputs 数组

对于输出数据集尚不存在的新数据流：

"outputs": [
  {
    "dataSourceId": null,
    "dataSourceName": "SFDC | Account Summary",
    "versionChainType": "REPLACE"
  }
]

将 dataSourceId 设为 null 会告诉 Domo 在首次运行时创建新数据集。首次成功运行后，Domo 会分配一个 UUID。

对于 已存在 的输出数据集：

"outputs": [
  {
    "dataSourceId": "8fe448eb-231d-4ff5-95f1-86db64336e96",
    "dataSourceName": "SFDC | Account Opportunity Summary",
    "versionChainType": "REPLACE"
  }
]

versionChainType："REPLACE" 会在每次运行时替换所有数据。其他选项包括 "APPEND"。

Action 类型

Action 是 DAG 中的节点。每个 action 都有唯一的 id、一个 type，并通过 dependsOn 引用其上游依赖项。

通用操作字段

每个操作都有以下字段：

{
  "type": "ActionType",
  "id": "unique-action-id",
  "name": "Human-readable name",
  "dependsOn": ["upstream-action-id-1", "upstream-action-id-2"],
  "settings": {
    "preferredDatabaseEntityType": "TEMP_VIEW"
  },
  "gui": {
    "x": 128,
    "y": 128,
    "color": null,
    "colorSource": null,
    "sampleJson": null
  },
  "tables": [{}]
}

字段	说明
`id`	此操作节点的唯一标识符。可以是任意字符串，但约定为 `TypeName-uuid`
`dependsOn`	必须在此操作运行前完成的操作 ID 数组
`gui.x` / `gui.y`	可视化画布中的位置。当 `useGraphUI: true` 时，顶层 `gui.canvases.default.elements` 数组优先——请同时设置两者以保持一致
`settings.preferredDatabaseEntityType`	始终为 `"TEMP_VIEW"`
`tables`	始终为 `[{}]`

LoadFromVault（输入节点）

将 Domo 数据集中的数据加载到 dataflow 中。

{
  "type": "LoadFromVault",
  "id": "LoadFromVault-accounts",
  "name": "RAW | SFDC | Accounts",
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 128, "y": 128, "color": null, "colorSource": null, "sampleJson": null},
  "previewRowLimit": 10000,
  "propagateAi": false,
  "filterPolicy": "LEGACY",
  "dataSourceId": "422efbf4-6c96-4576-907b-eacae8379d5d",
  "sourceType": "AUTO",
  "executeFlowWhenUpdated": false,
  "pseudoDataSource": false,
  "truncateTextColumns": false,
  "truncateRows": false,
  "onlyLoadNewVersions": false,
  "recentVersionCutoffMs": 0,
  "tables": [{}]
}

关键字段：

dataSourceId：要加载的数据集的 UUID
name：为清晰起见，应与数据集名称一致
无 dependsOn —— 这是一个根节点

MergeJoin（连接）

按指定的键连接两个上游操作。

{
  "type": "MergeJoin",
  "id": "MergeJoin-accounts-opps",
  "name": "Join Accounts & Opp Summary",
  "dependsOn": ["LoadFromVault-accounts", "GroupBy-opp-summary"],
  "disabled": false,
  "removeByDefault": false,
  "notes": [],
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 576, "y": 320, "color": null, "colorSource": null, "sampleJson": null},
  "previewRowLimit": null,
  "joinType": "LEFT OUTER",
  "relationshipType": "MTM",
  "step1": "LoadFromVault-accounts",
  "step2": "GroupBy-opp-summary",
  "keys1": ["Id"],
  "keys2": ["AccountId"],
  "on": null,
  "schemaModification1": [],
  "schemaModification2": [
    {"name": "AccountId", "rename": "Opp_AccountId", "remove": true}
  ]
}

字段	说明
`joinType`	`"LEFT OUTER"`、`"INNER"`、`"RIGHT OUTER"`、`"FULL OUTER"`
`relationshipType`	`"MTM"` —— 请使用此缩写，不要使用 `"MANY_TO_MANY"`（会导致验证错误）
`step1`	连接左侧的操作 ID
`step2`	连接右侧的操作 ID
`keys1`	来自 step1 的用于连接的列名数组
`keys2`	来自 step2 的用于连接的列名数组（按位置与 keys1 匹配）
`schemaModification1`	重命名或移除 step1（左侧）的列
`schemaModification2`	重命名或移除 step2（右侧）的列，以避免名称冲突
`dependsOn`	必须包含 `step1` 和 `step2` 两个操作 ID

注意： 键可以是不对称的——keys1: ["fldWorkCodeID"] / keys2: ["ID"]。Domo 按数组位置匹配，而不是按列名匹配。

schemaModification1 / schemaModification2

当两个数据集共享列名时（例如都含有 Id、Name、Description），使用 schemaModification2 来重命名或移除右侧（step2）数据集中的冲突列：

"schemaModification2": [
  {"name": "Description", "rename": "Opportunities.Description", "remove": false},
  {"name": "Id", "rename": "Opportunities.Id", "remove": false},
  {"name": "Name", "rename": "Opportunities.Name", "remove": false},
  {"name": "AccountId", "rename": "Opp_AccountId", "remove": true}
]

remove: true —— 将列从输出中完全删除
remove: false —— 保留该列，但将其重命名为 rename 的值

GroupBy（聚合）

通过分组列并应用聚合函数来聚合数据。

{
  "type": "GroupBy",
  "id": "GroupBy-opp-summary",
  "name": "Summarize Opportunities per Account",
  "dependsOn": ["LoadFromVault-opportunities"],
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 352, "y": 512, "color": null, "colorSource": null, "sampleJson": null},
  "input": "LoadFromVault-opportunities",
  "addLineNumber": false,
  "giveBackRow": false,
  "allRows": false,
  "groups": [
    {"name": "AccountId"}
  ],
  "fields": [
    {"name": "Opportunity Count", "source": "Id", "type": "COUNT_ALL", "valuefield": null, "expression": null, "settings": null},
    {"name": "Total Amount", "source": "Amount", "type": "SUM", "valuefield": null, "expression": null, "settings": null},
    {"name": "Avg Amount", "source": "Amount", "type": "AVERAGE", "valuefield": null, "expression": null, "settings": null},
    {"name": "Total Expected Revenue", "source": "ExpectedRevenue", "type": "SUM", "valuefield": null, "expression": null, "settings": null},
    {"name": "Avg Probability", "source": "Probability", "type": "AVERAGE", "valuefield": null, "expression": null, "settings": null}
  ],
  "tables": [{}]
}

字段	说明
`input`	上游数据的操作 ID
`groups`	用于分组的列数组。每一项：`{"name": "ColumnName"}`
`fields`	聚合定义数组

聚合字段定义

{
  "name": "Total Amount",
  "source": "Amount",
  "type": "SUM",
  "valuefield": null,
  "expression": null,
  "settings": null
}

字段	说明
`name`	聚合值的输出列名
`source`	要聚合的源列名
`type`	聚合函数（见下文）

聚合类型

类型	说明	已确认
`SUM`	值的总和	是
`AVERAGE`	值的平均值	是
`COUNT_ALL`	所有行的计数（包括空值）	是
`COUNT_DISTINCT`	不同值的计数
`MIN`	最小值	是
`MAX`	最大值
`FIRST`	组中的第一个值
`LAST`	组中的最后一个值
`CONCAT_COMMA`	用逗号分隔符连接值

重要： API 使用的是 MIN 和 MAX，而不是 MINIMUM 或 MAXIMUM。使用 MINIMUM 会返回验证错误："The value is invalid for expected type... AggregateType"。这是通过测试发现的。

基于表达式的聚合（高级）

除了使用 source + type，还可以使用完整的 SQL 聚合表达式。将 source、type 和 valuefield 设为 null，并将完整表达式放入 expression：

{
  "name": "Total Hours",
  "source": null,
  "type": null,
  "valuefield": null,
  "expression": "SUM(IFNULL(`Hours`,0))",
  "settings": null
}

关键： expression 必须是完整的 SQL 聚合表达式——而不仅仅是函数名。

"expression": "SUM(IFNULL(\Hours`,0))"搭配"source": null` —— 正确
"expression": "SUM" 搭配 "valuefield": "Hours" —— 错误（会把 "SUM" 当作列名）

这允许将 CASE WHEN 等复杂逻辑直接嵌入 GroupBy，而无需预先使用 GroupBy 前的 ExpressionEvaluator：

{
  "name": "LTM_Revenue",
  "source": null, "type": null, "valuefield": null,
  "expression": "SUM(CASE WHEN `Date` >= DATE_SUB(CURRENT_TIMESTAMP(), INTERVAL 12 MONTH) THEN IFNULL(`Revenue`,0) ELSE 0 END)",
  "settings": null
}

WindowAction（排名与窗口）

应用窗口函数——排名、行编号、lag/lead 偏移——按分组列分区并按排序列排序。这是 Magic ETL UI 中“Rank & Window”磁贴的编程等价物。

{
  "type": "WindowAction",
  "id": "WindowAction-first-order",
  "name": "First Order Date per Customer",
  "dependsOn": ["LoadFromVault-finance"],
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 352, "y": 128, "color": null, "colorSource": null, "sampleJson": null},
  "input": "LoadFromVault-finance",
  "groupRules": [
    {"column": "customer", "caseSensitive": false}
  ],
  "orderRules": [
    {"column": "date_ymd", "caseSensitive": false, "ascending": true}
  ],
  "additions": [
    {
      "name": "Customer Order Rank",
      "operation": {
        "type": "RANKING",
        "operationType": "ROW_NUMBER",
        "column": null,
        "defaultValue": null,
        "amount": null
      }
    }
  ],
  "tables": [{}]
}

字段	描述
`input`	上游数据的操作 ID
`groupRules`	分区列——窗口函数在每个分组内应用
`orderRules`	每个分区内的排序顺序
`additions`	要作为新列添加的窗口函数定义数组

groupRules

定义 PARTITION BY 列：

"groupRules": [
  {"column": "customer", "caseSensitive": false},
  {"column": "region", "caseSensitive": false}
]

每个条目按该列对数据分区。多个条目会创建复合分区键。

orderRules

定义每个分区内的 ORDER BY：

"orderRules": [
  {"column": "date_ymd", "caseSensitive": false, "ascending": true}
]

字段	描述
`column`	要排序的列
`caseSensitive`	字符串排序是否区分大小写
`ascending`	`true` 表示 ASC，`false` 表示 DESC

多个 orderRules 会创建复合排序键。

additions（窗口函数定义）

每个 addition 会创建一个包含窗口函数结果的新列：

{
  "name": "Output Column Name",
  "operation": {
    "type": "RANKING",
    "operationType": "ROW_NUMBER",
    "column": null,
    "defaultValue": null,
    "amount": null
  }
}

窗口操作类型

排名操作（type: "RANKING"）：

operationType	描述	column	amount
`ROW_NUMBER`	分区内的顺序编号（1, 2, 3...）	`null`	`null`
`RANK`	并列时留出间隔的排名（1, 2, 2, 4...）	`null`	`null`
`DENSE_RANK`	并列时不休间隔的排名（1, 2, 2, 3...）	`null`	`null`

偏移操作（type: "OFFSET"）：

operationType	描述	column	amount
`LAG`	分区中前 N 行的值	源列名	向前查看的行数
`LEAD`	分区中后 N 行的值	源列名	向后查看的行数

示例——使用 LAG 获取前一天的营收：

{
  "name": "Previous Day Revenue",
  "operation": {
    "type": "OFFSET",
    "operationType": "LAG",
    "column": "revenue",
    "defaultValue": null,
    "amount": 1
  }
}

示例——使用 ROW_NUMBER 对每位客户的订单进行排名：

{
  "name": "Customer Order Rank",
  "operation": {
    "type": "RANKING",
    "operationType": "ROW_NUMBER",
    "column": null,
    "defaultValue": null,
    "amount": null
  }
}

将 WindowAction 与 GroupBy 结合使用

一种常见模式是使用 WindowAction 进行行级排名，然后使用 GroupBy 进行聚合。例如，要查找每位客户的首个订单日期：

WindowAction：按 customer 分区，按 date_ymd ASC 排序，添加 ROW_NUMBER
GroupBy：按 customer 分组，对 date_ymd 使用 MIN 获取首个订单日期，并对数值列使用 SUM/AVERAGE

WindowAction 会传递所有原始列以及新增的 additions，因此下游操作可以访问所有内容。

Filter（行过滤）

根据列条件过滤行。这是 Magic ETL UI 中“Filter Rows”磁贴的编程等价物。

{
  "type": "Filter",
  "id": "Filter-exclude-florida",
  "name": "Exclude Florida",
  "dependsOn": ["LoadFromVault-sales"],
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 352, "y": 256, "color": null, "colorSource": null, "sampleJson": null},
  "input": "LoadFromVault-sales",
  "filterList": [
    {
      "leftField": "state",
      "rightField": null,
      "rightValue": {"value": "Florida", "type": "STRING"},
      "rightExpr": "'Florida'",
      "operator": "NE",
      "expression": null,
      "andFilterList": []
    }
  ],
  "tables": [{}]
}

字段	描述
`input`	上游数据的操作 ID
`filterList`	过滤条件数组（以 OR 逻辑组合）
`andFilterList`	过滤器内用于 AND 逻辑的嵌套数组

过滤条件字段

字段	描述
`leftField`	要过滤的列名
`operator`	比较运算符（见下表）
`rightValue`	带类型的值对象：`{"value": "Florida", "type": "STRING"}`
`rightExpr`	值的字符串表达式，字符串需用单引号包裹：`"'Florida'"`
`rightField`	要比较的列名（用于列与列比较，否则为 `null`）

重要： 对于字符串比较，rightValue 必须是带有 value 和 type 字段的带类型对象，并且 rightExpr 必须用单引号包裹该值。对 rightValue 使用普通字符串会返回 VALIDATION-DE 错误。

基于表达式的过滤器（替代方式）

除了使用 leftField/operator/rightValue，你还可以使用原始 SQL expression 来处理复杂条件：

{
  "leftField": null,
  "rightField": null,
  "rightValue": null,
  "rightExpr": null,
  "operator": null,
  "expression": "IFNULL(TRIM(`column_name`),'') <> ''",
  "andFilterList": []
}

这对于多条件过滤器或带函数调用的过滤器很有用。该表达式是 SQL WHERE 子句片段。

筛选运算符

运算符	描述	已确认
`NE`	不等于	是
`EQ`	等于
`LT`	小于
`GT`	大于
`LE`	小于或等于
`GE`	大于或等于
`NN`	非空	是（来自现有 dataflow）
`NL`	为空
`NIN`	不在列表中
`IN`	在列表中

rightValue 类型值

类型	用于
`STRING`	文本比较
`NUMERIC`	数字比较
`DATE`	日期比较

ExpressionEvaluator（添加公式）

使用 Domo 的表达式语言添加计算列。这是 Magic ETL UI 中“Add Formula”图块的程序化等效实现。

{
  "type": "ExpressionEvaluator",
  "id": "ExpressionEvaluator-month-trunc",
  "name": "Add Month Column",
  "dependsOn": ["Filter-exclude-florida"],
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 480, "y": 256, "color": null, "colorSource": null, "sampleJson": null},
  "input": "Filter-exclude-florida",
  "expressions": [
    {
      "expression": "concat(year(`date_ymd`),MONTHNAME(`date_ymd`))",
      "fieldName": "Year_Month",
      "settings": {}
    }
  ],
  "tables": [{}]
}

字段	描述
`input`	上游数据的 Action ID
`expressions`	公式定义数组

表达式定义

{
  "expression": "concat(year(`date_ymd`),MONTHNAME(`date_ymd`))",
  "fieldName": "Year_Month",
  "settings": {}
}

字段	描述
`expression`	Domo 公式表达式。列名必须用反引号包裹。
`fieldName`	新输出列的名称
`settings`	空对象 `{}`

已知可用函数

函数	描述	示例
`year()`	从日期中提取年份	`year(\`date_ymd`)`
`MONTH()`	提取月份数字（1-12）	`MONTH(\`date_ymd`)`
`MONTHNAME()`	从日期中提取月份名称	`MONTHNAME(\`date_ymd`)`
`concat()`	拼接字符串	`concat(year(\`date_ymd`), MONTHNAME(`date_ymd`))`
`DATE()`	转换为日期类型	`DATE(\`date_ymd`)`
`DATE_FORMAT()`	将日期格式化为字符串	`DATE_FORMAT(\`date_ymd`, 'yyyy-MM-dd')`
`LEFT()`	左侧子字符串	`LEFT(\`Date`, 7)` 用于从 “2025-01-15” 提取年月
`LPAD()`	左侧填充字符串	`LPAD(MONTH(\`date`), 2, '0')`
`IFNULL()`	空值替换（仅 2 个参数）	`IFNULL(\`col`, 0)`
`COALESCE()`	空值替换（多参数）	`COALESCE(\`a`, `b`, 'default')`
`TRIM()`	去除空白	`TRIM(\`col`)`
`CAST()`	类型转换	`CAST(NULL AS DATETIME)`
`LAST_DAY()`	当月最后一天	`LAST_DAY(\`date`)`
`DATE_SUB()`	减去时间间隔	`DATE_SUB(\`date`, INTERVAL 12 MONTH)`
`CONVERT_TZ()`	时区转换	`CONVERT_TZ(CURRENT_TIMESTAMP(),'UTC','US/Eastern')`
`CURRENT_TIMESTAMP()`	当前日期时间	`CURRENT_TIMESTAMP()`

重要： TRUNC_MONTH() 不是有效函数，会返回 "Unknown function: TRUNC_MONTH"。要将日期截断到月级别，请使用 concat(year(...), MONTHNAME(...)) 或类似的基于字符串的方法。

表达式行为规则

表达式按顺序处理——后面的表达式可以引用同一图块中前面表达式计算出的字段
添加与现有列具有相同 fieldName 的表达式会替换该列的值（适用于类型转换或重新格式化）
IFNULL() 仅接受 2 个参数；多参数空值处理请使用 COALESCE(a, b, c)
CAST(NULL AS DATETIME) 可用于创建带类型的空列

可以在单个 ExpressionEvaluator 操作中添加多个表达式——每个表达式都会创建一个新列（或替换现有列）。所有原始列都会透传。

Unique（去重）

根据指定的键列删除重复行。这是“Remove Duplicates”图块的程序化等效实现。

{
  "type": "Unique",
  "id": "Unique-dedup",
  "name": "Deduplicate",
  "dependsOn": ["SelectValues-final"],
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 800, "y": 256, "color": null, "colorSource": null, "sampleJson": null},
  "input": "SelectValues-final",
  "countRows": false,
  "fields": [
    {"name": "ColA", "caseInsensitive": false},
    {"name": "ColB", "caseInsensitive": false}
  ],
  "tables": [{}]
}

字段	描述
`input`	上游数据的 Action ID
`countRows`	为 `true` 时添加一个计数列，显示发现了多少重复项
`fields`	定义唯一性的列数组。在列出的所有列中具有相同值的行会被去重
`fields[].caseInsensitive`	为 `true` 时不区分大小写进行字符串匹配

提示： 在可能产生行扇出的连接（例如多对多连接）之后添加 Unique 图块。

SelectValues (列选择 / 重命名)

选择特定列并可选择对其进行重命名。

{
  "type": "SelectValues",
  "id": "SelectValues-final",
  "name": "Select Final Columns",
  "dependsOn": ["MergeJoin-final"],
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 1024, "y": 320, "color": null, "colorSource": null, "sampleJson": null},
  "input": "MergeJoin-final",
  "select": [
    {"name": "Id", "rename": "Account ID"},
    {"name": "Name", "rename": "Account Name"},
    {"name": "Industry"},
    {"name": "Contact Count"},
    {"name": "Total Amount"}
  ],
  "tables": [{}]
}

仅 select 中列出的列会被传递；其他所有列都会被丢弃
省略 rename 可保留原始列名

替代格式 — 使用 fields 代替 select。只列出你想要保留的列（可附带重命名）。未列出的列会被丢弃。不要添加 remove: true 条目 — 它们会导致 DP-0003 验证错误。

只列出你想要保留的列（可附带重命名）。未列出的列会被丢弃。不要添加 remove: true 条目 — 它们会导致 DP-0003 验证错误。

"fields": [
  {"name": "ID", "rename": "Dataflow ID"},
  {"name": "Display Name", "rename": "Dataflow Name"},
  {"name": "Link", "rename": "Dataflow Link"}
]

最小字段条目只需 {"name": "...", "rename": "..."}。不要包含 type、dateFormat、settings 或 remove — 在 fields 格式下这些会导致 DP-0003 Action is improperly configured 错误。

PublishToVault (输出节点)

将最终数据写入 Domo 数据集。

{
  "type": "PublishToVault",
  "id": "PublishToVault-output",
  "name": "SFDC | Account Summary",
  "dependsOn": ["MergeJoin-final"],
  "disabled": false,
  "removeByDefault": false,
  "notes": [],
  "settings": {"preferredDatabaseEntityType": "TEMP_VIEW"},
  "gui": {"x": 1248, "y": 320, "color": null, "colorSource": null, "sampleJson": null},
  "previewRowLimit": null,
  "dataSource": {
    "type": "DataFlow",
    "name": "SFDC | Account Summary",
    "cloudId": "domo"
  },
  "versionChainType": "REPLACE",
  "partitionIdColumns": [],
  "upsertColumns": [],
  "retainPartitionExpression": ""
}

字段	描述
`dependsOn`	包含单个操作 ID 的数组 — 最终的转换步骤
`dataSource.type`	`"DataFlow"`
`dataSource.name`	输出数据集的名称
`dataSource.guid`	现有输出数据集的 UUID — 对于新数据集请完全省略
`dataSource.cloudId`	`"domo"`
`versionChainType`	`"REPLACE"`（完全替换）或 `"APPEND"`
`partitionIdColumns`	`[]` — 除非使用分区，否则始终为空
`upsertColumns`	`[]` — 除非执行 upsert，否则始终为空
`retainPartitionExpression`	`""` — 始终为空字符串

关键： 不要在 PublishToVault 中包含 inputs、schemaSource、partitioned 或 tables。这些字段会导致 DP-DSCF 提交失败，此时 ETL 会成功处理所有行，但随后无法写入输出数据集。这一点已通过大量测试确认（2026 年 4 月）。

对于新的 dataflow，请完全省略 guid。Domo 会在首次成功运行时分配一个真实的输出数据集 UUID。

完整示例：账户汇总 ETL

此示例创建一个 dataflow，它会：

加载 3 个 Salesforce 数据集（Accounts、Contacts、Opportunities）
按账户聚合联系人（计数）
按账户聚合商机（计数、总金额、平均金额、预期收入、平均概率）
将两个聚合结果左连接到 accounts 表
为每个账户输出一行，包含所有汇总指标

流水线图

LoadFromVault(Accounts) ─────────────────────┐
                                              ├─ MergeJoin(LEFT) ──┐
LoadFromVault(Opportunities)                  │   on Id=AccountId   │
  └─ GroupBy(AccountId) ─────────────────────┘                     │
       COUNT(Id) → Opportunity Count                               ├─ MergeJoin(LEFT) ─── PublishToVault
       SUM(Amount) → Total Amount                                  │   on Id=AccountId
       AVG(Amount) → Avg Amount                                    │
       SUM(ExpectedRevenue) → Total Expected Revenue               │
       AVG(Probability) → Avg Probability                          │
                                                                   │
LoadFromVault(Contacts)                                            │
  └─ GroupBy(AccountId) ──────────────────────────────────────────┘
       COUNT(Id) → Contact Count

创建命令

curl -s -X POST "https://<instance>.domo.com/api/dataprocessing/v1/dataflows" \
  -H "Content-Type: application/json" \
  -H "x-domo-developer-token: <TOKEN>" \
  -d @sfdc_account_summary_etl.json

执行

# Via CLI
echo -e "connect -server <instance>.domo.com -token <TOKEN>\ndataflow-run-now -i <DATAFLOW_ID>\nquit" \
  | java -jar domoutil.jar

# Via API
curl -s -X POST "https://<instance>.domo.com/api/dataprocessing/v1/dataflows/<ID>/executions" \
  -H "x-domo-developer-token: <TOKEN>"

完整示例：带排名与窗口的客户汇总

此示例创建一个 dataflow，它会：

加载一个财务数据集（329,460 行，包含 customer、date、sales、revenue、cogs）
使用 WindowAction 按日期对每个客户的订单进行排名（ROW_NUMBER）
按客户聚合：首次订单日期（MIN）、sales/revenue/cogs 的 SUM 和 AVERAGE、订单计数

流水线图

LoadFromVault(Mododata | Finance)  329,460 rows
  └── WindowAction: ROW_NUMBER partitioned by customer, ordered by date_ymd ASC
       └── GroupBy(customer):
            MIN(date_ymd)  → First Order Date
            SUM(sales)     → Total Sales
            AVG(sales)     → Avg Sales
            SUM(revenue)   → Total Revenue
            AVG(revenue)   → Avg Revenue
            SUM(cogs)      → Total COGS
            AVG(cogs)      → Avg COGS
            COUNT(date_ymd)→ Order Count
              └── PublishToVault  →  104 customer rows

关键 JSON 片段

WindowAction（排名与窗口 tile）：

{
  "type": "WindowAction",
  "id": "WindowAction-first-order",
  "name": "First Order Date per Customer",
  "dependsOn": ["LoadFromVault-finance"],
  "input": "LoadFromVault-finance",
  "groupRules": [
    {"column": "customer", "caseSensitive": false}
  ],
  "orderRules": [
    {"column": "date_ymd", "caseSensitive": false, "ascending": true}
  ],
  "additions": [
    {
      "name": "Customer Order Rank",
      "operation": {
        "type": "RANKING",
        "operationType": "ROW_NUMBER",
        "column": null,
        "defaultValue": null,
        "amount": null
      }
    }
  ]
}

用于首次订单日期的带 MIN 的 GroupBy：

{
  "type": "GroupBy",
  "id": "GroupBy-customer-summary",
  "name": "Customer Aggregations",
  "dependsOn": ["WindowAction-first-order"],
  "input": "WindowAction-first-order",
  "groups": [{"name": "customer"}],
  "fields": [
    {"name": "First Order Date", "source": "date_ymd", "type": "MIN"},
    {"name": "Total Sales", "source": "sales", "type": "SUM"},
    {"name": "Avg Sales", "source": "sales", "type": "AVERAGE"},
    {"name": "Total Revenue", "source": "revenue", "type": "SUM"},
    {"name": "Avg Revenue", "source": "revenue", "type": "AVERAGE"},
    {"name": "Total COGS", "source": "cogs", "type": "SUM"},
    {"name": "Avg COGS", "source": "cogs", "type": "AVERAGE"},
    {"name": "Order Count", "source": "date_ymd", "type": "COUNT_ALL"}
  ]
}

结果

输出数据集：104 行（每个客户一行），包含列：customer、First Order Date、Total Sales、Avg Sales、Total Revenue、Avg Revenue、Total COGS、Avg COGS、Order Count。

完整示例：带筛选与公式的月度部门指标

此示例创建一个 dataflow，它会：

加载一个销售数据集（481,643 行）
筛选掉 state = 'Florida' 的行
使用 concat(year(), MONTHNAME()) 添加 Year_Month 列
按 Year_Month + department + metric_name 聚合：revenue、store_revenue、web_revenue、mobile_revenue、total_costs、visits 的 SUM 和 AVERAGE

流水线图

LoadFromVault(Mododata | Sales)  481,643 rows
  └── Filter: state != 'Florida'
       └── ExpressionEvaluator: Year_Month = concat(year(date_ymd), MONTHNAME(date_ymd))
            └── GroupBy(Year_Month, department, metric_name):
                 SUM/AVG for revenue, store_revenue, web_revenue,
                 mobile_revenue, total_costs, visits
                   └── PublishToVault  →  14,057 rows

关键 JSON 片段

Filter（排除特定值）：

{
  "type": "Filter",
  "id": "Filter-exclude-florida",
  "name": "Exclude Florida",
  "dependsOn": ["LoadFromVault-sales"],
  "input": "LoadFromVault-sales",
  "filterList": [
    {
      "leftField": "state",
      "rightField": null,
      "rightValue": {"value": "Florida", "type": "STRING"},
      "rightExpr": "'Florida'",
      "operator": "NE",
      "expression": null,
      "andFilterList": []
    }
  ]
}

ExpressionEvaluator（日期转换为年-月字符串）：

{
  "type": "ExpressionEvaluator",
  "id": "ExpressionEvaluator-month-trunc",
  "name": "Add Month Column",
  "dependsOn": ["Filter-exclude-florida"],
  "input": "Filter-exclude-florida",
  "expressions": [
    {
      "expression": "concat(year(`date_ymd`),MONTHNAME(`date_ymd`))",
      "fieldName": "Year_Month",
      "settings": {}
    }
  ]
}

结果

输出：14,057 行（每个月 x 部门 x metric_name 一行，排除 Florida）。粒度从 481K 每日行减少到 14K 月度聚合行。

ETL 错误代码参考

当数据流执行失败时，lastExecution.errors[] 中的错误对象包含一个 code 字段。使用此表进行诊断：

代码	类别	含义	典型修复方法
`DP-DSNF`	缺少输入	所需数据集不存在	重新创建数据集或移除 LoadFromVault 图块
`DP-61100`	缺少输入	与 DP-DSNF 相同（备用代码）	同上
`DP-0001`	架构不匹配	引用了但未找到列	上游架构已更改 — 更新列引用
`DP-0059`	公式错误	表达式中的语法错误	修复公式（例如，CASE 语句缺少 `END`）
`DP-0118`	代码/模型崩溃	转换作业失败	检查 Python 脚本或 AI 模型图块
`DP-TGTFR`	数据量	批量文本生成需要 >= 100 行	确保输入有足够的行数
`MYSQL-601072`	缺少连接键	表中不存在键列	更新连接键列名
`DP-0000`	未知	无 API 级别详细信息	在 Domo UI 中检查

将错误追踪到特定图块

错误对象包含一个 parameters 映射，其中带有 _actionId（失败图块的 UUID）。将其与数据流中的 actions[].id 进行匹配：

errors = dataflow.get("lastExecution", {}).get("errors", [])
actions_by_id = {a["id"]: a for a in dataflow.get("actions", [])}
for e in errors:
    action_id = e.get("parameters", {}).get("_actionId", "")
    tile = actions_by_id.get(action_id, {})
    print(f'Error: [{e["code"]}] {e["localizedMessage"]}')
    print(f'Tile: "{tile.get("name", "?")}" (type: {tile.get("type", "?")})')
    print(f'Property: {e.get("parameters", {}).get("_propertyPath", "")}')

`_actionId` 图块类型	需要检查的内容
`ExpressionEvaluator`	`expressions[].expression` — 公式文本
`LoadFromVault`	`dataSourceId` — 输入数据集 UUID
`MergeJoin`	`keys1` / `keys2` — 连接键列名
`SelectValues`	`fields[].name` — 被选择的列名
`Filter`	`filterList[].leftField` — 筛选列
`PublishToVault`	`dataSource.guid` — 输出数据集 UUID

常见陷阱

1. `DP-DSCF` 提交失败 — PublishToVault 结构错误

错误 "Failed to commit data to data source <UUID>"（代码 DP-DSCF）由不正确的 PublishToVault 字段引起。ETL 将成功处理所有行，但在提交步骤失败。

根本原因（2026 年 4 月确认）： 在 PublishToVault 操作中使用了 inputs、schemaSource、partitioned 或 tables。

修复方法： 使用此确切结构 — partitionIdColumns、upsertColumns、retainPartitionExpression — 并省略 inputs、schemaSource、partitioned、tables：

{
  "type": "PublishToVault",
  "dataSource": {"type": "DataFlow", "name": "Output Name", "cloudId": "domo"},
  "versionChainType": "REPLACE",
  "partitionIdColumns": [],
  "upsertColumns": [],
  "retainPartitionExpression": ""
}

2. SelectValues `fields` 格式 — 仅列出要保留的列

SelectValues 的 fields 格式会导致 DP-0003 Action is improperly configured，如果你包含带有 "remove": true 的条目或额外字段如 type、dateFormat、settings。仅列出你想要保留的列：

"fields": [
  {"name": "ID", "rename": "Dataset ID"},
  {"name": "Name", "rename": "Dataset Name"}
]

未列出的列会自动丢弃。不要添加 remove 条目。

3. 连接中的列名冲突

当连接两个共享列名（例如 Id、Name、Description）的数据集时，你必须在 MergeJoin 上使用 schemaModification2 来重命名或移除右侧数据集中的冲突列。未能这样做会导致下游出现不明确的列名。

4. 操作 ID 必须唯一

actions 数组中的每个操作都需要一个唯一的 id 字符串。约定是 TypeName-uuid（例如 "LoadFromVault-b71d89d7-6c08-44ad-a503-ebed046377e0"），但任何唯一字符串都可以（例如 "LoadFromVault-accounts"）。

5. dependsOn 必须匹配 step1/step2/input

对于 MergeJoin，dependsOn 数组必须包含 step1 和 step2 两个操作 ID。对于 GroupBy，dependsOn 必须包含 input 操作 ID。不匹配将导致数据流失败。

6. GUI 位置影响可视化画布

gui.x 和 gui 的值决定节点在 Domo Magic ETL 可视化编辑器中的显示位置。为了让布局整齐，节点之间大约保持水平 224px、垂直 192px 的间距，输入节点通常从 x: 128 开始。

当设置了 useGraphUI: true（现代画布模式）时，图块位置由顶层的 gui.canvases.default.elements 数组决定，而不是由操作级别的 gui 字段决定。为了保持一致，始终要同时设置两者。完整的 Section 和 Tile 元素 schema 请参阅“顶层 gui 字段”一节。

7. 通过 API 创建 Dataflow — 认证（2026 年 4 月更新）

认证： 在大多数实例上，POST /api/dataprocessing/v1/dataflows 可使用开发者令牌（DDCI...）（2026 年 4 月已确认）。请求体必须包含 "databaseType": "MAGIC"。创建的 dataflow 不处于 DRAFT 状态，可以立即通过 API 执行——无需在 UI 中保存。

输出： 当 POST 包含 "outputs": [{"dataSourceId": null, "dataSourceName": "...", "versionChainType": "REPLACE"}] 时，Domo 会立即分配一个真实的输出数据集。第一次成功执行时会创建实际的数据集。

CLI： domo（ryuu）CLI 仅用于 Custom App 发布，没有 dataflow 命令。CLI（domoutil.jar）有 dataflow 命令，但需要单独设置。

8. 聚合类型名称和你想的不一样

GroupBy 的 type 字段使用 MIN 和 MAX，而不是 MINIMUM 或 MAXIMUM。使用 MINIMUM 会返回验证错误：

{"code":"VALIDATION-DIV","message":"The value is invalid for expected type.","path":"actions[2].fields[0].type","parameters":{"field":"type","type":"AggregateType","rejected":"MINIMUM"}}

已确认可用的类型：SUM、AVERAGE、COUNT_ALL、4。请使用这些确切的字符串。

9. Filter 的 right 必须是类型化对象

Filter 操作的 rightValue 不能是普通字符串，必须是类型化对象：

// WRONG — returns VALIDATION-DE error
"rightValue": "Florida"

// CORRECT
"rightValue": {"value": "Florida", "type": "STRING"}

此外，rightExpr 必须用单引号包裹字符串值："'Florida'"。rightValue 和 rightExpr 应一起设置。

10. TRUNC_MONTH 及其他日期函数并不存在

ExpressionEvaluator 不支持 TRUNC_MONTH()、DATE_TRUNC() 或类似的日期截断函数。要获取年月值，请使用字符串拼接：

concat(year(`date_ymd`), MONTHNAME(`date_ymd`))

已知可用的函数：year()、MONTHNAME()、concat()、DATE()。

11. 通过 POST 创建的 Dataflow 可以立即执行（2026 年 4 月更新）

与之前的说明相反，使用开发者令牌通过 POST /api/dataprocessing/v1/dataflows 创建的 dataflow 不处于 DRAFT 状态，可以立即执行通过 POST .../executions。当传入 dataSourceId: null 时，创建时 outputs 数组会被填充。无需在 UI 中保存。执行会返回有效的执行 ID 并运行至完成。

12. CLI dataflow-run-now 可能在新 Dataflow 上失败

Java CLI 的 dataflow-run-now 命令对于新建的 dataflow 可能会返回 500 错误。请始终使用 API 来运行（POST /api/dataprocessing/v1/dataflows/{id}/executions）——它更可靠。

13. responsibleUserId 是可选的

无需在 POST 请求体中指定 responsibleUserId——它默认使用已认证用户。

14. LoadFromVault 节点可以供给多个分支

单个 LoadFromVault 节点可以供给多个下游分支——只需在多个 dependsOn 数组中引用其 ID 即可。这样可以避免重复加载同一数据集。

15. PublishToVault — 仅使用 `dependsOn`，不要使用 `inputs`

PublishToVault 操作使用 dependsOn 来引用其上游操作。不要包含 inputs 数组——它不是可用格式的一部分，并且会导致 DP-DSCF 提交失败。如果创建后 numOutputs = 0，则很可能是操作中缺少 dataSource 对象——用 PUT 提交修正后的定义来修复。

16. API 令牌认证

REST API 使用与 CLI 相同的开发者令牌：

x-domo-developer-token: <TOKEN>

这与 OAuth 不同——不需要交换客户端 ID/密钥。

19. 所有非 LoadFromVault 操作都需要 `disabled`、`removeByDefault`、`notes`

除 LoadFromVault 外，每个操作都必须包含这些字段，否则数据流在 UI 中可能会出现意外行为：

"disabled": false,
"removeByDefault": false,
"notes": [],
"previewRowLimit": null

MergeJoin 还需要 "on": null，以及显式的 "schemaModification1": [] 和 "schemaModification2": []（即使为空也需要）。

LoadFromVault 使用 "previewRowLimit": 10000（不是 null）。

20. `relationshipType` 必须是 `"MTM"`，而不是 `"MANY_TO_MANY"`

MergeJoin 字段 relationshipType 必须使用缩写形式 "MTM"。使用 "MANY_TO_MANY" 会导致校验错误。其他有效值："OTO"、"OTM"、"MTO"。

21. 构建血缘 ETL 前始终询问主数据集锚点

在构建任何数据集血缘 ETL 之前，先问：“是否有一个主 ‘My Datasets’ 或类似的列表数据集，每个输出行都应关联回它？” 这个锚点数据集（通常包含 Dataset ID、Name、Link、Owner 等）应作为最终 join 的 LEFT 侧加载，以便所有数据集都出现在输出中——即使那些没有 card/ETL 关联的独立数据集。缺少这个锚点意味着输出只覆盖恰好出现在其他元数据表中的数据集，会遗漏任何独立数据集。

17. 获取数据流时不要使用 `?hydrate=full`

GET /api/dataprocessing/v1/dataflows/{id}?hydrate=full 对某些数据流会返回 400 Bad Request。非 hydrated 的端点（GET /api/dataprocessing/v1/dataflows/{id}）已经返回完整的 actions[] 数组以及所有 tile 详情。始终使用非 hydrated 版本。

18. PUT 会替换整个数据流定义

当通过 PUT /api/dataprocessing/v1/dataflows/{id} 更新数据流时：

发送完整的数据流对象 — PUT 会替换整个定义。始终从一次新的 GET 开始。
PUT 只会保存 — 它不会运行数据流。请通过 POST .../executions 单独触发执行。
版本跟踪 — 每次成功的 PUT 都会递增 onboardFlowVersion.versionNumber。执行响应包含 dataFlowVersion，用于确认运行的是哪个版本。

已知操作类型

实例中现有数据流中发现的所有操作类型：

操作类型	描述	已确认可用
`LoadFromVault`	输入节点 — 加载一个 Domo 数据集	是
`PublishToVault`	输出节点 — 写入一个 Domo 数据集	是
`MergeJoin`	基于键 join 两个上游操作	是
`GroupBy`	使用 SUM、AVERAGE、MIN、COUNT_ALL 等进行聚合	是
`WindowAction`	Rank & Window — ROW_NUMBER、RANK、LAG、LEAD	是
`SelectValues`	选择并重命名列	是（但在 PublishToVault 之前可能导致提交错误）
`Filter`	基于条件过滤行（NE、EQ、NN 等）	是
`ExpressionEvaluator`	添加计算列 / 公式（concat、year、MONTHNAME、DATE）	是
`Unique`	按键列去重行	已记录（来自模式指南）
`Metadata`	更改列类型或元数据	在现有数据流中见过
`SplitColumnAction`	将一列拆分为多列	在现有数据流中见过
`FixedInput`	硬编码 / 常量输入数据	在现有数据流中见过

工作流：创建新的 Magic ETL

飞行前问题（构建前询问）

所有输入数据集 ID 及其列 schema 是什么？（通过 /api/query/v1/execute/<UUID> 使用 SELECT * FROM table LIMIT 2 逐个查询）
是否有一个主 / 锚点数据集（例如 “My Datasets”），每个输出行都应关联回它？如果有，获取它的 UUID——它将成为最终 join 的 LEFT 侧。
期望的输出列名具体是什么，以及每个输入字段映射到哪个输出列？

构建步骤

获取输入数据集 schema — 对每个输入数据集执行 POST /api/query/v1/execute/<UUID>，请求体为 {"sql": "SELECT * FROM table LIMIT 2"}
构建 JSON 定义 — 使用本文档中已确认可用的 action 结构
添加带颜色的 Section 区域 — 将相关的 tile 分组到 gui.canvases.default.elements 中的 Section 元素。为每个逻辑分支或处理阶段分配一种独特的颜色。使用 parentId 和相对坐标将 Tile 元素重新挂载到各自的 section 中。对于所有具有 2 个及以上分支或阶段的数据流，此步骤为必需。
通过 API 创建 — 使用 dev token 调用 POST /api/dataprocessing/v1/dataflows — 立即生效，无需在 UI 中保存
执行 — POST /api/dataprocessing/v1/dataflows/<ID>/executions
检查状态 — GET /api/dataprocessing/v1/dataflows/<ID>/executions/<EXEC_ID> — 轮询直到 state 为 SUCCESS 或 FAILED_DATA_FLOW
失败时 — 检查执行响应中的 errors[] 数组，查找 actionId 和 localizedMessage，修复具体的 action，PUT 更新后的定义，然后重新运行
导出以进行调试 — GET /api/dataprocessing/v1/dataflows/<ID> 返回完整保存的定义，包括服务器分配的所有 GUID

支持 Markdown

登录后评论

基本信息

作者: stahura
仓库: stahura/domo-ai-vibe-rules
分类: 数据与数据库
安装次数: 178
GitHub Stars: 24

报告问题

magic-etl

关于 magic-etl

主要功能

适用场景

安装

详细介绍

在 Domo 中以编程方式创建 Magic ETL 数据流

概述

数据流的 CLI 命令

列出所有 Magic ETL 数据流

获取数据流定义（导出为 JSON）

运行数据流

列出数据流执行记录

通过 API 检查执行状态

通过 API 创建数据流

端点

更新现有数据流

数据流 JSON 结构

顶层字段

顶层 gui 字段（画布布局与分区）

Section 元素（彩色区域背景）

可用的 Section 背景颜色

Tile 元素（操作位置）

创建 Dataflow 时务必添加 Section

Inputs 数组

Outputs 数组

Action 类型

通用操作字段

LoadFromVault（输入节点）

MergeJoin（连接）

schemaModification1 / schemaModification2

GroupBy（聚合）

聚合字段定义

聚合类型

基于表达式的聚合（高级）

WindowAction（排名与窗口）

groupRules

orderRules

additions（窗口函数定义）

窗口操作类型

将 WindowAction 与 GroupBy 结合使用

Filter（行过滤）

过滤条件字段

基于表达式的过滤器（替代方式）

筛选运算符

rightValue 类型值

ExpressionEvaluator（添加公式）

表达式定义

已知可用函数

表达式行为规则

Unique（去重）

SelectValues (列选择 / 重命名)

PublishToVault (输出节点)

完整示例：账户汇总 ETL

流水线图

创建命令

执行

完整示例：带排名与窗口的客户汇总

流水线图

关键 JSON 片段

结果

完整示例：带筛选与公式的月度部门指标

流水线图

关键 JSON 片段

结果

ETL 错误代码参考

将错误追踪到特定图块

常见陷阱

1. DP-DSCF 提交失败 — PublishToVault 结构错误

2. SelectValues fields 格式 — 仅列出要保留的列

3. 连接中的列名冲突

4. 操作 ID 必须唯一

5. dependsOn 必须匹配 step1/step2/input

6. GUI 位置影响可视化画布

7. 通过 API 创建 Dataflow — 认证（2026 年 4 月更新）

8. 聚合类型名称和你想的不一样

9. Filter 的 right 必须是类型化对象

10. TRUNC_MONTH 及其他日期函数并不存在

11. 通过 POST 创建的 Dataflow 可以立即执行（2026 年 4 月更新）

12. CLI dataflow-run-now 可能在新 Dataflow 上失败

顶层 `gui` 字段（画布布局与分区）

1. `DP-DSCF` 提交失败 — PublishToVault 结构错误

2. SelectValues `fields` 格式 — 仅列出要保留的列

15. PublishToVault — 仅使用 `dependsOn`，不要使用 `inputs`

19. 所有非 LoadFromVault 操作都需要 `disabled`、`removeByDefault`、`notes`

20. `relationshipType` 必须是 `"MTM"`，而不是 `"MANY_TO_MANY"`

17. 获取数据流时不要使用 `?hydrate=full`