ScriptedTransformRecord

说明

ScriptedTransformRecord 提供了使用脚本语言（例如 Groovy）快速、 方便地更新记录内容的能力。Lynxus 提供了多个不同的 Processor， 可以用不同方式操作记录，每个 Processor 各有优劣。 ScriptedTransformRecord 可能是功能最强大、最通用的选择。但它也 最容易出错，因为它依赖编写自定义脚本。它的性能也可能最低， 因为直接用 Java 编写的 Processor 和库通常比解释型脚本性能更好。

编写脚本时需要注意，与 ExecuteScript 不同，该 Processor 不允许 脚本本身暴露可配置的属性或定义 Relationship。这是刻意的决定。 如果需要暴露这类配置，应改用 ExecuteScript。不暴露这些元素， 脚本就无需定义类或实现具有特定方法签名的方法，可以完全避免 样板代码，专注于手头的任务。

对于传入 FlowFile 中遇到的每条记录，所提供的脚本都会被求值一次。 每次调用脚本时，期望它返回一个 Record 对象（关于返回值见下面的 说明）。该记录随后会使用配置的 Record Writer 写入。如果脚本 返回 null 值，该记录不会被写入。如果脚本返回的对象不是 Record，传入 FlowFile 会被路由到 failure Relationship。

该 Processor 维护两个计数器："Records Transformed" 表示传递给 脚本且脚本返回了 Record 的记录数量，"Records Dropped" 表示 传递给脚本且脚本返回了 null 值的记录数量。

变量绑定

虽然提供给该 Processor 的脚本不需要提供样板代码或实现任何 类/接口，但它需要某种方式来访问记录和其他所需信息。这通过 变量绑定实现。每次调用脚本时，以下每个变量都会提供给脚本：

变量名	说明	变量类型
record	要转换的记录。	Record
recordIndex	记录在 FlowFile 中从 0 开始的索引。	Long（64 位带符号整数）
log	Processor 的日志器。写入该日志器的任何内容都会写入日志，就好像 Processor 自己记录的一样。此外，写入该日志器的任何日志消息都会创建一条公告（bulletin）（默认情况下，Processor 会隐藏级别低于 WARN 的公告）。	ComponentLog
attributes	FlowFile Attribute 的键/值对 Map。该 Map 的键和值都是 String 类型。该 Map 是不可变的，任何修改尝试都会抛出 UnsupportedOperationException。	java.util.Map

返回值

每次调用脚本时，期望它返回一个 Record 对象或 Record 对象的集合。这些记录随后会使用配置的 Record Writer 写入。如果脚本返回 null 值，该记录不会被 写入。如果脚本返回的对象不是 Record 或 Record 集合，传入 FlowFile 会被路由到 failure Relationship。

提供给脚本的记录是可变的。因此，常见的模式是在脚本中更新 record 对象，然后直接返回同一个 record 对象。

注意： 根据脚本语言的不同，没有显式返回值的脚本可能 返回 null，也可能返回最后引用的值。由于返回 null 会 导致记录被丢弃，返回非 Record 值会导致异常（也为了清晰 起见），确保配置的脚本有显式的返回值很重要。

添加新字段

面向记录的 Processor 的一个非常常见的用法是让 Record Reader 推断其 schema，并让 Record Writer 继承该记录的 schema。在这种场景下，需要注意的是，Record Writer 会继承 它遇到的第一条记录的 schema。因此，如果配置的脚本会给记录 添加新字段，确保该字段被添加到所有记录（适当时值为 null）很重要。

更多细节参见添加新字段示例。

性能注意事项

Lynxus 提供了许多不同的 Processor，以各种方式更新记录。 虽然各有优劣，但性能通常是重要的考量。一般来说，像 UpdateRecord 这样的标准 Processor 会比脚本类 Processor 性能更好。但情况未必总是如此。在性能关键的 场景下，最好的办法是测试两种方式，看哪种表现更好。

做了一个简单的 5 分钟基准测试来分析性能差异。所用脚本 只是修改一个字段，其余部分原样返回记录。结果如下。 注意：未给出硬件细节，因为这些结果不应被用来推断绝对 性能预期，而只是展示不同选项之间的相对性能。

Processor	所用脚本	5 分钟处理的记录数
UpdateAttribute	无脚本。添加名为 /num、值为 42 的用户定义属性	5010 万
ScriptedTransformRecord - 使用语言：Groovy	record.setValue("num", 42) record	1890 万

脚本示例

移除第一条记录

以下脚本会移除它遇到的每个 FlowFile 中的第一条记录。

输入示例（CSV）：

name, num Mark, 42 Felicia, 3720 Monica, -3

输出示例（CSV）：

name, num Felicia, 3720 Monica, -3

脚本示例（Groovy）：

return recordIndex == 0 ? null : record

替换字段值

以下脚本会在记录中某字段的值等于 "Value To Replace" Attribute 的值时，替换该字段。该字段的值会被替换为 "Replacement Value" Attribute 中的值。

输入内容示例（JSON）：

[
  {
    "book": {
      "author": "John Doe",
      "date": "01/01/1980"
    }
  },
  {
    "book": {
      "author": "Jane Doe",
      "date": "01/01/1990"
    }
  }
]

输入 Attribute 示例：

Attribute 名称	Attribute 值
Value To Replace	Jane Doe
Replacement Value	Author Unknown

输出示例（JSON）：

[
  {
    "book": {
      "author": "John Doe",
      "date": "01/01/1980"
    }
  },
  {
    "book": {
      "author": "Author Unknown",
      "date": "01/01/1990"
    }
  }
]

脚本示例（Groovy）：

def replace(rec) {
    rec.toMap().each { k, v ->
        // 如果字段值等于 Attribute 'Value to Replace'，则把
        // 字段值设置为 'Replacement Value' Attribute。
        if (v?.toString()?.equals(attributes['Value to Replace'])) {
            rec.setValue(k, attributes['Replacement Value'])
        }

        // 如果值是 Record，递归调用
        if (v instanceof cn.org.acal.lynxus.serialization.record.Record) {
            replace(v)
        }
    }
}

replace(record)
return record

直通（Pass-through）

以下脚本允许每条记录原样通过，不做任何修改。

输入示例：<任意>

输出示例：<与输入相同>

脚本示例（Groovy）：

record

添加新字段

以下脚本为所有记录添加一个名为 "favoriteColor" 的新字段。 此外，还为所有偶数索引的记录添加 "isOdd" 字段。

所有记录具有相同的 schema 很重要。由于我们想给记录 1 和 3 添加 "isOdd" 字段，记录 0 和 2 的 schema 也必须为此留出 位置。因此，我们会给所有记录添加该字段，但对于非偶数记录 使用 null 值。更多信息参见添加新字段。

输入内容示例（CSV）：

name, favoriteFood John Doe, Spaghetti Jane Doe, Pizza Jake Doe, Sushi June Doe, Hamburger

输出示例（CSV）：

name, favoriteFood, favoriteColor, isOdd John Doe, Spaghetti, Blue, Jane Doe, Pizza, Blue, true Jake Doe, Sushi, Blue, June Doe, Hamburger, Blue, true

脚本示例（Groovy）：

import cn.org.acal.lynxus.serialization.record.RecordField;
import cn.org.acal.lynxus.serialization.record.RecordFieldType;

// 始终把 favoriteColor 设为 Blue。
// 因为调用 #setValue 时字段名是 String，字段类型会被推断。
record.setValue("favoriteColor", "Blue")

// 如果记录索引为奇数，把 'isOdd' 字段设为 true；否则设为 `null`。
// 由于第一条记录的值可能是 `null`（这个特定场景下总是如此），
// 我们需要确保 Record Writer 的 schema 获得该字段的正确类型。
// 因此，不用 String 字段名调用 #setValue，而是传入一个
// RecordField 作为第一个参数，因为 RecordField 允许指定字段类型。
// 另外注意，`RecordField` 和 `RecordFieldType` 已在上面 `import`。
record.setValue(new RecordField("isOdd", RecordFieldType.BOOLEAN.getDataType()), recordIndex % 2 == 1 ? true : null)

return record

记录分叉（Fork Record）

以下脚本返回它遇到的每条记录，外加另一条从第一条派生的 记录，其 'num' 字段比输入的 'num' 字段小 1。

输入示例（CSV）：

name, num Mark, 42 Felicia, 3720 Monica, -3

输出示例（CSV）：

name, num Mark, 42 Mark, 41 Felicia, 3720 Felicia, 3719 Monica, -3 Monica, -4

脚本示例（Groovy）：

import cn.org.acal.lynxus.serialization.record.*

def derivedValues = new HashMap(record.toMap())
derivedValues.put('num', derivedValues['num'] - 1)
derived = new MapRecord(record.schema, derivedValues)
return [record, derived]