ExecuteScript

说明

ExecuteScript Processor 提供了使用脚本语言借助 Lynxus API 执行任务的 能力，例如：

• 从传入 FlowFile 读取内容和/或 Attribute
• 创建新的 FlowFile（有无父对象均可）
• 向输出 FlowFile 写入内容和/或 Attribute
• 与 ProcessSession 交互，把 FlowFile 转移到 Relationship
• 读写 State Manager，在 Processor 的多次执行之间跟踪变量

注意：

• ExecuteScript 使用 JSR-223 脚本引擎 API 对脚本求值，因此对地道 的语言结构有时支持有限。例如对于 Groovy，另有一个 ExecuteGroovyScript Processor，允许执行更多地道的 Groovy 任务。 例如，通过 ExecuteGroovyScript 与 Controller Service 交互比 ExecuteScript 更容易（详见 ExecuteGroovyScript 文档）。

变量绑定

该 Processor 期望由用户定义的脚本在 Processor 被触发时求值。以下 变量可供脚本使用：

变量名	说明	变量类型
session	这是分配给该 Processor 的 ProcessSession 的引用。通过 session 可以对 FlowFile 执行 create()、putAttribute()、transfer() 等操作，以及 read() 和 write()	ProcessSession
context	这是该 Processor 的 ProcessContext 的引用。可用它获取 Processor 属性、Relationship、Controller Service 和 State Manager。	ProcessContext
log	这是该 Processor 的 ComponentLog 的引用。用它向日志记录消息，例如 log.info('Hello world!')	ComponentLog
REL_SUCCESS	这是为该 Processor 定义的 "success" Relationship 的引用。它也可以通过引用父类（ExecuteScript）的静态成员继承，这里作为便捷变量提供，也避免了使用 Relationship 的全限定名。	Relationship
REL_FAILURE	这是为该 Processor 定义的 "failure" Relationship 的引用。与 REL_SUCCESS 一样，它也可以通过引用父类（ExecuteScript）的静态成员继承，这里作为便捷变量提供，也避免了使用 Relationship 的全限定名。	Relationship
动态属性	ExecuteScript 中定义的任何动态（用户自定义）属性都会作为变量传递给脚本引擎，变量设置为对应动态属性的 PropertyValue 对象。这样你可以获取属性的 String 值，也可以用 Lynxus 表达式语言对属性求值、把值转换为合适的数据类型（例如 Boolean）等。由于动态属性名会成为脚本的变量名，必须注意所选脚本引擎的变量命名规则。例如，Groovy 不允许变量名中包含句点（.），因此如果动态属性名为 "my.property" 会报错。与这些变量的交互通过 Lynxus Java API 完成。下面的"动态属性"小节会在引入时讨论相关的 API 调用。	PropertyValue

脚本示例

从 session 获取一个传入 FlowFile

用例：你有指向 ExecuteScript 的传入连接，想从队列中取出一个 FlowFile 进行处理。

方法：使用 session 对象的 get() 方法。该方法返回下一个优先级 最高的待处理 FlowFile。如果没有待处理的 FlowFile，方法返回 null。注意：即使有稳定的 FlowFile 流入 Processor，也可能返回 null。当 Processor 有多个并发任务且其他任务已取走 FlowFile 时， 会发生这种情况。如果脚本需要 FlowFile 才能继续，那么当 session.get() 返回 null 时应立即返回。

Groovy

flowFile = session.get()
if (!flowFile) return

从 session 获取多个传入 FlowFile：

用例：你有指向 ExecuteScript 的传入连接，想从队列中取出多个 FlowFile 进行处理。

方法：使用 session 对象的 get(maxResults) 方法。该方法从工作 队列中返回最多 maxResults 个 FlowFile。如果没有可用的 FlowFile，返回空列表（该方法不会返回 null）。注意：当存在多个 传入队列时，单次调用是轮询所有队列还是只轮询单个队列的行为是 未指定的。不过，观察到的行为（Lynxus 1.1.0 及之前版本）描述在 这里。

示例：

Groovy

flowFileList = session.get(100)
if (!flowFileList.isEmpty()) {
    flowFileList.each { flowFile ->
// 在这里处理每个 FlowFile
    }
}

创建新的 FlowFile

用例：你想生成一个新的 FlowFile 发送给下一个 Processor。

方法：使用 session 对象的 create() 方法。该方法返回一个新的 FlowFile 对象，你可以对其做进一步处理。

示例：

Groovy

flowFile = session.create()
// 在这里进行额外处理

从父 FlowFile 创建新的 FlowFile

用例：你想基于一个传入 FlowFile 生成新的 FlowFile。

方法：使用 session 对象的 create(parentFlowFile) 方法。该方法 接收一个父 FlowFile 引用，返回一个新的子 FlowFile 对象。新创建的 FlowFile 会继承父对象的所有 Attribute（UUID 除外）。该方法会根据 在 ProcessSession 提交之前是否从同一父对象生成了其他 FlowFile， 自动生成 Provenance FORK 事件或 Provenance JOIN 事件。

示例：

Groovy

flowFile = session.get()
if (!flowFile) return
newFlowFile = session.create(flowFile)
// 在这里进行额外处理

向 FlowFile 添加 Attribute

用例：你有一个 FlowFile，想为其添加自定义 Attribute。

方法：使用 session 对象的 putAttribute(flowFile, attributeKey, attributeValue) 方法。该方法用给定的键/值对更新给定 FlowFile 的 Attribute。注意："uuid" Attribute 对于 FlowFile 是固定的，不能修改；如果键名为 "uuid"，会被忽略。

另外这里值得一提的是，FlowFile 对象是不可变的；这意味着如果你 通过 API 更新 FlowFile 的 Attribute（或做其他修改），会得到指向 新版本 FlowFile 的新引用。在把 FlowFile 转移到 Relationship 时 这一点非常重要。你必须保留对最新版本 FlowFile 的引用，并且必须 转移或移除从 session 获取或创建的所有 FlowFile 的最新版本， 否则执行时会出错。大多数情况下，存储 FlowFile 引用的变量会被 修改 FlowFile 的方法返回的最新版本覆盖（中间的 FlowFile 引用会被自动丢弃）。在这些示例中，你会看到添加 Attribute 时 复用 FlowFile 引用的技巧。注意，当前对 FlowFile 的引用会被 传入 putAttribute() 方法。得到的 FlowFile 有一个名为 'myAttr'、值为 'myValue' 的 Attribute。另外，该方法接收 String 类型的值；如果你有一个 Object，需要先序列化为 String。最后请注意，如果要添加多个 Attribute，最好创建 一个 Map 并使用 putAllAttributes()（详见下一个配方）。

示例：

Groovy

flowFile = session.get()
if (!flowFile) return
flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')

向 FlowFile 添加多个 Attribute

用例：你有一个 FlowFile，想为其添加多个自定义 Attribute。

方法：使用 session 对象的 putAllAttributes(flowFile, attributeMap) 方法。该方法用给定 Map 中的键/值对更新给定 FlowFile 的 Attribute。注意："uuid" Attribute 对于 FlowFile 是固定的，不能修改；如果键名为 "uuid"，会被忽略。

示例：

Groovy

attrMap = ['myAttr1': '1', 'myAttr2': Integer.toString(2)]
flowFile = session.get()
if (!flowFile) return
flowFile = session.putAllAttributes(flowFile, attrMap)

从 FlowFile 获取 Attribute

用例：你有一个 FlowFile，想查看它的某个 Attribute。

方法：使用 FlowFile 对象的 getAttribute(attributeKey) 方法。 该方法返回给定 attributeKey 的 String 值，如果 attributeKey 不存在则返回 null。示例展示了获取 "filename" Attribute 的值。

示例：

Groovy

flowFile = session.get()
if (!flowFile) return
myAttr = flowFile.getAttribute('filename')

从 FlowFile 获取所有 Attribute

用例：你有一个 FlowFile，想获取它的所有 Attribute。

方法：使用 FlowFile 对象的 getAttributes() 方法。该方法返回 一个键和值都是 String 的 Map，表示该 FlowFile 的 Attribute 键/值对。示例展示了遍历 FlowFile 所有 Attribute 的 Map。

示例：

Groovy

flowFile = session.get()
if (!flowFile) return
flowFile.getAttributes().each { key, value ->
// 对这个键/值对做些处理
}

把 FlowFile 转移到 Relationship

用例：处理完一个 FlowFile（新的或传入的）后，你想把它 转移到某个 Relationship（"success" 或 "failure"）。在这个简单 案例中，假设有一个名为 "errorOccurred" 的变量，指示 FlowFile 应转移到哪个 Relationship。更多错误处理技术会在 本系列的第 2 部分讨论。

方法：使用 session 对象的 transfer(flowFile, relationship) 方法。文档中写道：该方法根据给定的 Relationship 把给定 FlowFile 转移到相应目标 Processor 的工作队列。如果 Relationship 指向多个目标，会复制 FlowFile 的状态，使每个 目标都收到 FlowFile 的完全相同的副本，尽管各自有唯一的 身份标识。

注意：ExecuteScript 会在每次执行结束时执行 session.commit()， 以确保操作已提交。你不需要（也不应该）在脚本中执行 session.commit()。

示例：

Groovy

flowFile = session.get()
if (!flowFile) return
// 在这里进行处理
if (errorOccurred) {
    session.transfer(flowFile, REL_FAILURE)
} else {
    session.transfer(flowFile, REL_SUCCESS)
}

以指定日志级别向日志发送消息

用例：你想把处理过程中发生的某个事件报告给日志框架。

方法：使用 log 变量的 warn()、trace()、debug()、info() 或 error() 方法。这些方法可以接收单个 String，或一个 String 加一个 Object 数组，或一个 String 加一个 Object 数组再加一个 Throwable。第一种用于简单消息。第二种用于想记录某些动态 对象/值时。要在消息字符串中引用它们，在消息中使用 "{}"。这些占位符会按出现顺序与 Object 数组求值，因此如果 消息为 "Found these things: {} {} {}"，Object 数组为 ['Hello',1,true]，记录的消息就是 "Found these things: Hello 1 true"。第三种形式的日志方法 还接收 Throwable 参数，在捕获到异常并想记录时有用。

示例：

Groovy

log.info('Found these things: {} {} {}', ['Hello', 1, true] as Object[])

使用回调读取传入 FlowFile 的内容

用例：你有指向 ExecuteScript 的传入连接，想从队列中获取 FlowFile 的内容进行处理。

方法：使用 session 对象的 read(flowFile, inputStreamCallback) 方法。需要一个 InputStreamCallback 对象传入 read() 方法。注意， 因为 InputStreamCallback 是一个对象，其内容默认只对该对象 可见。如果你需要在 read() 方法之外使用数据，请使用作用域更广 的变量。示例会把传入 FlowFile 的完整内容存入一个 String （使用 Apache Commons 的 IOUtils 类）。注意：对于大 FlowFile，这不是最佳技术；更好的做法是只读取所需的数据并 相应处理。对于类似 SplitText 的场景，你可以逐行读取并在 InputStreamCallback 内处理，或使用前面提到的 session.read(flowFile) 方式获取 InputStream 引用，在回调 之外使用。

示例：

Groovy

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets

flowFile = session.get()
if (!flowFile) return
def text = ''
// 把带 inputStream 参数的闭包转换为 InputStreamCallback
session.read(flowFile, { inputStream ->
    text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
// 在这里对 text 做些处理
} as InputStreamCallback)

使用回调向输出 FlowFile 写入内容

用例：你想为输出 FlowFile 生成内容。

方法：使用 session 对象的 write(flowFile, outputStreamCallback) 方法。需要一个 OutputStreamCallback 对象传入 write() 方法。 注意，因为 OutputStreamCallback 是一个对象，其内容默认 只对该对象可见。如果你需要在 write() 方法之外使用数据， 请使用作用域更广的变量。示例会向 FlowFile 写入一个示例 String。

示例：

Groovy

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets

flowFile = session.get()
if (!flowFile) return
def text = 'Hello world!'
// 把带 outputStream 参数的闭包转换为 OutputStreamCallback
flowFile = session.write(flowFile, { outputStream ->
    outputStream.write(text.getBytes(StandardCharsets.UTF_8))
} as OutputStreamCallback)

使用回调以更新后的内容覆盖传入 FlowFile

用例：你想复用传入 FlowFile，但想修改其输出内容。

方法：使用 session 对象的 write(flowFile, streamCallback) 方法。需要一个 StreamCallback 对象传入 write() 方法。 StreamCallback 同时提供一个 InputStream（来自传入 FlowFile）和一个 outputStream（用于该 FlowFile 的下一个 版本），因此你可以用 InputStream 获取 FlowFile 的当前 内容，修改后再写回该 FlowFile。这会覆盖 FlowFile 的内容， 因此对于追加（append），你需要通过追加到读入的内容来 处理，或使用不同的方式（用 session.append() 而不是 session.write()）。注意，因为 StreamCallback 是一个 对象，其内容默认只对该对象可见。如果你需要在 write() 方法之外使用数据，请使用作用域更广的变量。 示例会把传入 FlowFile 的内容（假设为 String）反转，并把 反转后的字符串写入 FlowFile 的新版本。

示例：

Groovy

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets

flowFile = session.get()
if (!flowFile) return
def text = 'Hello world!'
// 把带 inputStream 和 outputStream 参数的闭包转换为 StreamCallback
flowFile = session.write(flowFile, { inputStream, outputStream ->
    text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
    outputStream.write(text.reverse().getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
session.transfer(flowFile, REL_SUCCESS)

处理脚本处理过程中的错误

用例：脚本中发生错误（通过数据验证或抛出异常），你想让 脚本优雅地处理它。

方法：对于异常，使用脚本语言的异常处理机制（通常是 try/catch 块）。对于数据验证，你可以使用类似的方式，但 定义一个布尔变量（如 "valid"）和 if/else 子句，而不是 try/catch 子句。ExecuteScript 定义了 "success" 和 "failure" Relationship；通常你的处理会把"好的" FlowFile 转移到 success，把"坏的"转移到 failure （后者会记录错误日志）。

示例：

Groovy

flowFile = session.get()
if (!flowFile) return
try {
// 这里可能会抛出异常

// 最后一个操作是转移到 success（失败在 catch 块中处理）
    session.transfer(flowFile, REL_SUCCESS)
} catch (e) {
    log.error('Something went wrong', e)
    session.transfer(flowFile, REL_FAILURE)
}