UpdateAttribute

说明

该 Processor 使用用户添加的属性或规则更新 FlowFile 的 Attribute。使用该 Processor 添加或修改 Attribute 有三种方式。第一种是"基本用法"：可以设置影响每个经过 Processor 的 FlowFile 的默认 Attribute 变更。第二种是"高级用法"：可以进行条件式 Attribute 变更，只有当 FlowFile 满足特定条件时才生效。可以在同一个 Processor 中同时使用这两种方式。第三种是 "Delete Attributes Expression"：可以提供一个正则表达式， 名称匹配的 Attribute 都会被删除。

请注意，"Delete Attributes Expression" 优先于任何更新。如果 已有 Attribute 匹配 "Delete Attributes Expression"，无论它是否被更新都会被移除。 也就是说，"Delete Attributes Expression" 只适用于输入 FlowFile 中已存在的 Attribute；如果 Attribute 是由该 Processor 添加的，"Delete Attributes Expression" 不会检测到它。

属性：

该 Processor 中的属性由用户添加。用户添加的属性支持表达式语言。 如何编写正确的表达式语言语句以执行所需功能，参见 Lynxus 表达式语言指南。

如果添加的 Attribute 名为 alternate.identifier 且其值是 URI，会注册一个 ADD_INFO Provenance 事件，把 FlowFile 与给定的备用标识符关联起来。

Relationship：

• success
  • 如果 Processor 成功更新了指定的 Attribute，FlowFile 会走这个 Relationship。
• set state fail
  • 如果 Processor 以有状态方式运行，在向 FlowFile 添加 Attribute 后设置状态失败，FlowFile 会被路由到这个 Relationship。

基本用法

基本用法通过添加新的 Processor 属性来进行变更，属性名引用你想修改的 Attribute。然后把期望的 Attribute 值输入为 Value。Value 可以是简单的任意文本字符串，也可以是指定如何构造值的 Lynxus 表达式语言语句。（编写 Lynxus 表达式语言语句的详情参见 Lynxus 表达式语言使用指南。）

例如，要修改标准的 "filename" Attribute，在末尾追加 ".txt"，就添加一个新属性，把属性名设为 "filename"（引用目标 Attribute），值使用下面所示的 Lynxus 表达式语言语句：

• 属性：filename
• 值：${filename}.txt

上面的示例说明了如何修改已有 Attribute。如果 Attribute 尚不存在，该 Processor 也可以用来添加新 Attribute。例如，可以添加以下属性，创建一个名为 myAttribute、值为 myValue 的新 Attribute：

• 属性：myAttribute
• 值：myValue

本例中，所有经过该 Processor 的 FlowFile 都会获得一个额外的名为 myAttribute、值为 myValue 的 FlowFile Attribute。这种配置可以用在想给每个 FlowFile 打标签以便在流程后面（例如在 RouteOnAttribute Processor 中路由）使用的场景。

高级用法

前面的示例说明了如何对每个经过 Processor 的 FlowFile 进行变更。不过，UpdateAttribute Processor 也可以进行条件式变更。

要根据某些条件修改 Attribute，在画布上下文菜单中点击 Advanced 菜单项，使用 Processor 的高级用户界面（UI）。

点击 Advanced 菜单项会显示高级 UI。在高级 UI 中，条件及其关联的动作以"规则（Rules）"形式输入。每条规则 基本上表示"如果满足这些条件，就执行这个动作"。一条规则可以使用 一个或多个条件，所有条件都必须满足才会执行指定的动作。

添加规则

要添加第一条规则，点击屏幕中央的 "Create Rule" 按钮。会显示 Edit Rule 表单，可以输入规则的名称、注释、条件和动作。规则 定义完成后，点击 Add 按钮。可以点击规则列表右上角带加号的按钮添加更多规则。

规则示例

本例有两条规则：CheckForLargeFiles 和 CheckForGiantFiles。CheckForLargeFiles 规则有以下条件：

• ${filename:equals('fileOfInterest')}
• ${fileSize:toNumber():ge(1048576)}
• ${fileSize:toNumber():lt(1073741824)}

然后对 filename Attribute 有这个动作：

• ${filename}.meg

综合起来，这条规则表示：

• 如果 filename Attribute 的值是 fileOfInterest，并且
• 如果 fileSize 大于等于（ge）1 兆字节（1,048,576 字节），并且
• 如果 fileSize 小于（lt）1 吉字节（1,073,741,824 字节）
• 那么通过在文件名末尾追加 ".meg" 来修改 filename Attribute 的值。

添加另一条规则

继续这个示例，我们可以再添加一条规则，检查大于 1 吉字节的文件。添加第二条规则时，可以利用要克隆的规则菜单中的 "Clone Rule" 选项，把之前的规则作为模板。这会打开预填充了 现有规则条件的新规则表单。

本例中，CheckForGiantFiles 规则有以下条件：

• ${filename:equals('fileOfInterest')}
• ${fileSize:toNumber():gt(1073741824)}

然后对 filename Attribute 有这个动作：

• ${filename}.gig

综合起来，这条规则表示：

• 如果 filename Attribute 的值是 fileOfInterest，并且
• 如果 fileSize 大于（gt）1 吉字节（1,073,741,824 字节）
• 那么通过在文件名末尾追加 ".gig" 来修改 filename Attribute 的值。

组合基本用法与高级用法

UpdateAttribute Processor 允许同时进行基本用法变更（即对每个 FlowFile）和高级用法变更（即条件式）；不过，如果两者影响相同的 Attribute，条件式变更优先。这还有一个额外的好处，即支持一种 "else" 结构。换句话说，如果没有任何规则匹配该 Attribute，就会应用基本用法变更。

删除 Attribute

删除 Attribute 很简单，只需提供要删除的 Attribute 名称的正则表达式。这可以是匹配单个 Attribute 的简单正则表达式，也可以是匹配一组同名 Attribute 甚至多个独立 Attribute 名称的更复杂正则表达式。

• lastUser - 会删除名为 "lastUser" 的 Attribute。
• user.* - 会删除以 "user" 开头的 Attribute，例如 "username、"userName"、"userID" 和 "users"。但不会删除 "User" 或 "localuser"。
• (user.*|host.*|.*Date) - 会删除 "user"、"username"、"userName"、"hostInfo"、"hosts" 和 "updateDate"，但不会删除 "User"、"HOST"、"update" 或 "updatedate"。

如果删除了 alternate.identified Attribute，删除 Attribute 功能不会产生 Provenance 事件。

FlowFile 策略

高级 UI 中的另一个设置是 FlowFile 策略。它位于 UI 左上角，定义了多条规则匹配时 Processor 的行为。可以用滑动开关切换。默认情况下， FlowFile 策略设为对每条匹配的规则使用原始 FlowFile 的克隆。

如果 FlowFile 策略设为 "use clone"，且多条规则匹配，会创建传入 FlowFile 的副本，使输出 FlowFile 的数量等于匹配规则的数量。换句话说，如果两条规则 （A 和 B）都匹配，会有两个输出 FlowFile，一个对应规则 A，一个对应规则 B。在想添加 Attribute 作为后续路由标志的场景下很有用。本例中会有文件的 两个副本可用，一个路由到 A 路径，一个路由到 B 路径。

如果 FlowFile 策略设为 "use original"，所有匹配的规则会应用于同一个传入 FlowFile，只有一个输出 FlowFile，应用了所有 Attribute 变更。此时规则的顺序很重要，每条匹配规则的动作会按 该顺序应用。如果多条规则包含更新同一 Attribute 的动作，最后一条匹配规则的动作优先。值得注意的是， 当 FlowFile 策略设为 "use original" 且用户切换了 "Reorder rules" 控件后，可以在规则列表中拖放规则来调整顺序。 在这种重排序模式下，不允许进行其他规则修改。

过滤规则

高级 UI 支持创建任意多的规则。为了管理大型规则集，可以使用左下角的 Filter 机制过滤规则列表。规则可以按名称、条件或动作中的任意文本 过滤。

关闭高级 UI

在高级 UI 中保存所有变更后，可以使用顶部的导航返回画布。

有状态用法

为 "Store State" 属性选择 "store state locally" 选项后，UpdateAttribute 不仅会把求值后的属性存储为 FlowFile 的 Attribute，还会存储为有状态变量，以递归方式引用。这使得 Processor 可以计算诸如传入 FlowFile 的总和或计数。动态属性可以这样作为有状态变量引用：

• 动态属性
  • 键：theCount
  • 值：${getStateValue("theCount"):plus(1)}

本例会持续统计经过该 Processor 的 FlowFile 总数。要在状态之上使用逻辑，只需使用 UpdateAttribute 的"高级用法"。所有动作都会存储为有状态 Attribute，同时也会添加到 FlowFile。使用"高级用法"可以跟踪诸如流程到目前为止的最大值。 做法是设置条件 "${getStateValue("maxValue"):lt(${value})}" 和动作 attribute:"maxValue"、value:"${value}"。"Stateful Variables Initial Value" 属性用于初始化有状态变量，有状态运行时必须设置。某些逻辑规则 需要非常高的初始值，例如使用高级规则确定最小值。如果有状态属性 引用其他有状态属性，其他有状态属性的值会落后一次迭代。例如， 计算传入流的平均值需要总和与计数。如果三个属性都设置在同一个 UpdateAttribute 中（如下），Average 将始终不包含 count 和 sum 的最新值：

• Count
  • 键：theCount
  • 值：${getStateValue("theCount"):plus(1)}
• Sum
  • 键：theSum
  • 值：${getStateValue("theSum"):plus(${flowfileValue})}
• Average
  • 键：theAverage
  • 值：${getStateValue("theSum"):divide(getStateValue("theCount"))}

作为替代，由于 average 只依赖 theCount 和 theSum Attribute（它们也会添加到 FlowFile），应该在其后接一个无状态的 UpdateAttribute 来正确计算平均值。如果 Processor 在 onTrigger 开始时无法获取状态，FlowFile 会被推回源 Relationship，且 Processor 会让出（yield）。如果 Processor 在 onTrigger 开始时能获取状态，但在向 FlowFile 添加 Attribute 后无法设置状态，FlowFile 会被转移到 "set state fail"。这通常是因为状态不是最新版本（另一个线程 已用另一个版本替换了状态）。在大多数用例中，这个 Relationship 应该循环回 Processor，因为只有受影响的 Attribute 会被覆盖。注意：目前唯一的"有状态"选项是本地存储 状态。这是因为当前集群状态实现依赖 Zookeeper，而 Zookeeper 不是为 UpdateAttribute 带状态所需的负载/吞吐量类型设计的。未来， 如果/当添加了多种不同的集群状态选项时，UpdateAttribute 会相应更新。

组合高级用法与有状态

UpdateAttribute Processor 允许同时使用高级用法变更（即条件式） 和把值存储到状态。这使得 UpdateAttribute 可以充当有状态规则引擎，实现有限状态机或跟踪 最小/最大值等强大概念。两者配合使用相对简单：当 Processor 本来要更新 Processor 上的 Attribute 时（即匹配了条件规则），同样的更新会存储到状态。 通过高级标签页引用状态的方式也相同，使用 "getStateValue"。注意：如果设置了 "use clone" 策略且设置状态失败，不会生成克隆，只有原始 FlowFile 会被转移到 "set state fail"。

关于并发与有状态用法的说明

使用有状态选项时，应谨慎使用并发任务。如果每个传入 FlowFile 都会更新状态，只用一个任务会高效得多。这是因为 onTrigger 做的第一件事是获取状态，最后一件事是（如果有更新） 存储状态。如果更新时它没有最新的初始状态，会失败并把 FlowFile 发送到 "set state fail"。这样做是为了确保更新是基于最新信息完成的。 如果不以这种模拟原子的方式进行，就无法保证状态是准确的。考虑 并发时，用例通常分为三类：

• 每个 FlowFile 都更新状态的数据流，例如更新计数器
• FlowFile 不总是更新状态的数据流，例如有限状态机
• 不更新状态的数据流，加上一个每次都更新但相对数据流很稀少的 第二"控制"流，例如触发器

第一种和最后一种情况的指导相对明确。对于第一种，不应使用并发。 这样做只会浪费 CPU，并发的任何好处都会因状态未命中而被抵消。对于最后一种， 可以轻松使用并发。由于更新本来就稀少，两个更新同时处理导致 问题的情况更加罕见。第二种情况有些灰色地带。如果更新稀少， 可能可以使用并发。如果更新频繁，并发可能带来的问题多于 好处。无论如何，通过测试确定合适的调优是唯一可靠的