ForkRecord Processor

类型全限定名：cn.org.acal.lynxus.processors.standard.ForkRecord

所属 Bundle：cn.org.acal.lynxus:lynxus-standard-nar:2.11.8

描述

该处理器允许用户将一条记录分叉为多条记录。用户必须至少指定一个记录路径作为动态属性，指向包含 RECORD 对象的 ARRAY 类型字段。处理器接受两种模式：“拆分”和“提取”。在这两种模式下，指定数组中包含的每个元素都会生成一条记录。在“拆分”模式下，每个生成的记录将保留与输入中给定的模式相同的模式，但数组将仅包含一个元素。在“提取”模式下，数组的元素必须是记录类型，并且将是生成的记录。此外，在“提取”模式下，可以指定每个生成的记录是否应包含从根级别到提取记录的父记录的所有字段。这假设要在记录中添加的字段是在Record WriterController Service的架构中定义的。请参阅该处理器的附加详细信息文档中的示例。

属性

下表中，加粗的属性为必填项；其余为可选项。表格同时标注默认值、是否支持表达式语言，以及是否为敏感属性（值会被加密）。

名称	默认值	可选值	描述
Record Reader		控制器服务 API： `RecordReaderFactory` cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8	指定用于读取传入数据的 Controller Service
Record Writer		控制器服务 API： `RecordSetWriterFactory` cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8	指定用于写出记录的 Controller Service
Mode	`split`	提取 — 生成的记录将是数组的元素拆分 — 生成的记录将保留输入模式并包含一个单元素数组	指定处理器的分叉模式
Include Parent Fields	`false`	true false	此参数仅在 'extract' 模式下有效。如果设置为 true，从根级别到给定数组的所有字段都将作为数组中每个元素的字段添加，以便进行分叉。

动态属性

动态属性允许用户自定义属性的名称与值。

名称	值	描述
Record Path 属性	Record Path 的值	一个 Record Path 值，指向包含 RECORD 对象的 ARRAY 类型字段

关系 (Relationships)

名称	描述	自动终止
`failure`	如果 FlowFile 在 fork 操作期间生成错误，它将被路由到此关系	否
`fork`	包含分叉记录的 FlowFiles 将被路由到此关系	否
`original`	原始的FlowFiles将被路由到这个关系	否

写入的 FlowFile 属性

名称	描述
`record.count`	生成的 FlowFile 将具有 'record.count' 属性，指示写入 FlowFile 的记录数。
`mime.type`	Record Writer 指示的 MIME 类型
`<Attributes from Record Writer>`	配置的记录写入器返回的任何属性都将添加到 FlowFile 中。

状态管理

该组件不存储状态。

受限 (Restricted)

该组件不受限。

输入要求

该组件要求有传入连接。

详细说明

ForkRecord

ForkRecord 允许用户把一条记录分叉（fork）成多条记录。为此，用户必须指定一个或多个 RecordPath（作为 Processor 的动态属性），指向一个元素类型为 RECORD 的 ARRAY 字段。

该 Processor 支持两种模式：

Split 模式 - 在此模式下，生成的记录与输入具有相同的 schema。数组中的每个元素会生成一条记录，且数组只包含该元素。
Extract 模式 - 在此模式下，生成的记录就是数组中包含的元素。此外，还可以把从根层级到被分叉记录元素的所有父记录字段添加到每条记录中。不过这要求要添加的字段在 Record Writer Controller Service 的 schema 中有定义。

示例

EXTRACT 模式

为了更好地理解该 Processor 的工作方式，我们给出几个示例。为便于说明，假设输入数据是 JSON 格式，内容如下：

[
  {
    "id": 1,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "accounts": [
      {
        "id": 42,
        "balance": 4750.89
      },
      {
        "id": 43,
        "balance": 48212.38
      }
    ]
  },
  {
    "id": 2,
    "name": "Jane Doe",
    "address": "345 My Street",
    "city": "Her City",
    "state": "NY",
    "zipCode": "22222",
    "country": "USA",
    "accounts": [
      {
        "id": 45,
        "balance": 6578.45
      },
      {
        "id": 46,
        "balance": 34567.21
      }
    ]
  }
]

示例 1 - 不带父字段提取

在这种情况下，我们希望为每个 account 创建一条记录，不关心其他字段。我们添加一个值为 /accounts 的动态属性 "path"。结果 FlowFile 将包含 4 条记录，内容如下（假设 Record Writer 的 schema 已正确设置）：

[
  {
    "id": 42,
    "balance": 4750.89
  },
  {
    "id": 43,
    "balance": 48212.38
  },
  {
    "id": 45,
    "balance": 6578.45
  },
  {
    "id": 46,
    "balance": 34567.21
  }
]

示例 2 - 带父字段提取

如果把 "Include parent fields" 属性设为 true，则会递归地把父字段包含到输出记录中（前提是 Record Writer 的 schema 允许）。如果多个字段同名（如本例中的 id），子字段优先于所有同名的父字段。在本例中，数组 accounts 的 id 会保存在分叉后的记录中。结果 FlowFile 将包含 4 条记录，内容如下：

[
  {
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "id": 42,
    "balance": 4750.89
  },
  {
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "id": 43,
    "balance": 48212.38
  },
  {
    "name": "Jane Doe",
    "address": "345 My Street",
    "city": "Her City",
    "state": "NY",
    "zipCode": "22222",
    "country": "USA",
    "id": 45,
    "balance": 6578.45
  },
  {
    "name": "Jane Doe",
    "address": "345 My Street",
    "city": "Her City",
    "state": "NY",
    "zipCode": "22222",
    "country": "USA",
    "id": 46,
    "balance": 34567.21
  }
]

示例 3 - 多层嵌套数组

假设输入记录包含多层嵌套数组，如下例：

[
  {
    "id": 1,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "accounts": [
      {
        "id": 42,
        "balance": 4750.89,
        "transactions": [
          {
            "id": 5,
            "amount": 150.31
          },
          {
            "id": 6,
            "amount": -15.31
          }
        ]
      },
      {
        "id": 43,
        "balance": 48212.38,
        "transactions": [
          {
            "id": 7,
            "amount": 36.78
          },
          {
            "id": 8,
            "amount": -21.34
          }
        ]
      }
    ]
  }
]

如果我们希望每个 account 的每个 transaction 各生成一条记录，那么 Record Path 应设为 /accounts[*]/transactions。如果 Record Reader 的 schema 如下：

{
  "type": "record",
  "name": "bank",
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "address",
      "type": "string"
    },
    {
      "name": "city",
      "type": "string"
    },
    {
      "name": "state",
      "type": "string"
    },
    {
      "name": "zipCode",
      "type": "string"
    },
    {
      "name": "country",
      "type": "string"
    },
    {
      "name": "accounts",
      "type": {
        "type": "array",
        "items": {
          "type": "record",
          "name": "accounts",
          "fields": [
            {
              "name": "id",
              "type": "int"
            },
            {
              "name": "balance",
              "type": "double"
            },
            {
              "name": "transactions",
              "type": {
                "type": "array",
                "items": {
                  "type": "record",
                  "name": "transactions",
                  "fields": [
                    {
                      "name": "id",
                      "type": "int"
                    },
                    {
                      "name": "amount",
                      "type": "double"
                    }
                  ]
                }
              }
            }
          ]
        }
      }
    }
  ]
}

且 Record Writer 的 schema 如下：

{
  "type": "record",
  "name": "bank",
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "address",
      "type": "string"
    },
    {
      "name": "city",
      "type": "string"
    },
    {
      "name": "state",
      "type": "string"
    },
    {
      "name": "zipCode",
      "type": "string"
    },
    {
      "name": "country",
      "type": "string"
    },
    {
      "name": "amount",
      "type": "double"
    },
    {
      "name": "balance",
      "type": "double"
    }
  ]
}

那么，如果包含父字段，会得到如下 4 条记录：

[
  {
    "id": 5,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "amount": 150.31,
    "balance": 4750.89
  },
  {
    "id": 6,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "amount": -15.31,
    "balance": 4750.89
  },
  {
    "id": 7,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "amount": 36.78,
    "balance": 48212.38
  },
  {
    "id": 8,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "amount": -21.34,
    "balance": 48212.38
  }
]

SPLIT 模式

示例

假设我们有如下数据，并添加了值为 /accounts 的属性 "path"：

[
  {
    "id": 1,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "accounts": [
      {
        "id": 42,
        "balance": 4750.89
      },
      {
        "id": 43,
        "balance": 48212.38
      }
    ]
  },
  {
    "id": 2,
    "name": "Jane Doe",
    "address": "345 My Street",
    "city": "Her City",
    "state": "NY",
    "zipCode": "22222",
    "country": "USA",
    "accounts": [
      {
        "id": 45,
        "balance": 6578.45
      },
      {
        "id": 46,
        "balance": 34567.21
      }
    ]
  }
]

则会得到如下 4 条记录：

[
  {
    "id": 1,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "accounts": [
      {
        "id": 42,
        "balance": 4750.89
      }
    ]
  },
  {
    "id": 1,
    "name": "John Doe",
    "address": "123 My Street",
    "city": "My City",
    "state": "MS",
    "zipCode": "11111",
    "country": "USA",
    "accounts": [
      {
        "id": 43,
        "balance": 48212.38
      }
    ]
  },
  {
    "id": 2,
    "name": "Jane Doe",
    "address": "345 My Street",
    "city": "Her City",
    "state": "NY",
    "zipCode": "22222",
    "country": "USA",
    "accounts": [
      {
        "id": 45,
        "balance": 6578.45
      }
    ]
  },
  {
    "id": 2,
    "name": "Jane Doe",
    "address": "345 My Street",
    "city": "Her City",
    "state": "NY",
    "zipCode": "22222",
    "country": "USA",
    "accounts": [
      {
        "id": 46,
        "balance": 34567.21
      }
    ]
  }
]

ForkRecord

ForkRecord Processor

描述

标签

属性

动态属性

关系 (Relationships)

写入的 FlowFile 属性

状态管理

受限 (Restricted)

输入要求

详细说明

ForkRecord

示例

EXTRACT 模式

示例 1 - 不带父字段提取

示例 2 - 带父字段提取

示例 3 - 多层嵌套数组

SPLIT 模式

示例