JoinEnrichment

简介

JoinEnrichment Processor 设计用于与 ForkEnrichment Processor 配合。两者结合使用， 提供了一套强大的机制：先将数据转换为用于获取富化（enrichment）数据的独立请求负载， 获取富化数据，可选地对富化数据进行转换，最后将原始负载与富化数据合并。

典型数据流

ForkEnrichment Processor 负责接收一个 FlowFile 并产生它的两个副本：一个送往 "original" Relationship，另一个送往 "enrichment" Relationship。每个副本都会被 追加一组各自的 Attribute。

"original" FlowFile 被路由到 JoinEnrichment Processor，而 "enrichment" FlowFile 被路由到另一个方向。这两个 FlowFile 都会带有一个名为 "enrichment.group.id"、 取值相同的 Attribute。JoinEnrichment Processor 随后利用该信息将两个 FlowFile 关联起来。每个 FlowFile 还会被追加 "enrichment.role" Attribute， 但取值不同：路由到 "original" 的 FlowFile 的 enrichment.role 为 ORIGINAL， 路由到 "enrichment" 的 FlowFile 的 enrichment.role 为 ENRICHMENT。

构成 "enrichment" 路径的 Processor 因用例而异。在本例中，我们使用 JoltTransformJSON Processor 把原始负载转换成 Web 服务所期望的请求负载， 再使用 InvokeHTTP Processor 获取与用例相关的富化数据。该路径上常用的其他 Processor 还包括 QueryRecord、UpdateRecord、ReplaceText、JoltTransformRecord 和 ScriptedTransformRecord。使用上述一个或多个 Processor 对通过 InvokeHTTP 调用得到的 Web 服务响应进行转换，也是一种常见的用法。

获取到富化数据之后，如果不能把它和原始负载合并起来，这些数据就没什么用处。 为此使用 JoinEnrichment Processor，它负责合并来自 "original" FlowFile 和 "enrichment" FlowFile 的记录。

JoinEnrichment Processor 为 "original" FlowFile 和 "enrichment" FlowFile 分别配置独立的 RecordReader。这意味着原始数据和富化数据可以拥有完全不同的 schema，甚至可以使用不同的数据格式。例如，原始负载可以是 CSV 数据，而 富化数据是 JSON 负载。由于使用了 RecordReader，这完全没有问题。该 Processor 还需要一个 RecordWriter，用于写出富化后的负载（即合并了 "original" 和 "enrichment" 数据的负载）。

JoinEnrichment Processor 提供了多种把原始记录与富化数据合并的策略， 下面逐一详细说明。

Wrapper（包裹）

Wrapper 策略是默认策略。原始负载中的每条记录都应在富化负载中有一条对应记录。 输出记录是包含 original 和 enrichment 两个字段的记录，二者分别包含 对应 FlowFile 中的第 n 条记录。例如，如果原始 FlowFile 的内容如下：

id, name, age
28021, John Doe, 55
832, Jane Doe, 22
29201, Jake Doe, 23
555, Joseph Doe, 2

富化 FlowFile 的内容如下：

id, email
28021, john.doe@lynxus.acal.org.cn
832, jane.doe@lynxus.acal.org.cn
29201, jake.doe@lynxus.acal.org.cn

该策略会产生如下输出（假设使用 JSON Writer）：

[
  {
    "original": {
      "id": 28021,
      "name": "John Doe",
      "age": 55
    },
    "enrichment": {
      "id": 28021,
      "email": "john.doe@lynxus.acal.org.cn"
    }
  },
  {
    "original": {
      "id": 832,
      "name": "Jane Doe",
      "age": 22
    },
    "enrichment": {
      "id": 832,
      "email": "jane.doe@lynxus.acal.org.cn"
    }
  },
  {
    "original": {
      "id": 29201,
      "name": "Jake Doe",
      "age": 23
    },
    "enrichment": {
      "id": 29201,
      "email": "jake.doe@lynxus.acal.org.cn"
    }
  },
  {
    "original": {
      "id": 555,
      "name": "Joseph Doe",
      "age": 2
    },
    "enrichment": null
  }
]

使用此策略时，原始 FlowFile 的第一条记录与富化 FlowFile 的第一条记录配对， 第二条与第二条配对，以此类推。如果某个 FlowFile 的记录数多于另一个， 缺失的位置使用 null 值。

Insert Enrichment Fields（插入富化字段）

"Insert Enrichment Fields" 策略把 "enrichment" 记录的所有字段插入原始记录。 记录按其在 FlowFile 中的索引关联：即 "enrichment" FlowFile 的第一条记录 插入 "original" FlowFile 的第一条记录，第二条插入第二条，以此类推。

选择此策略时，"Record Path" 属性是必需的。Record Path 会对 "original" 记录求值。例如，考虑 "original" FlowFile 的以下内容：

[
  {
    "purchase": {
      "customer": {
        "loyaltyId": 48202,
        "firstName": "John",
        "lastName": "Doe"
      },
      "total": 48.28,
      "items": [
        {
          "itemDescription": "book",
          "price": 24.14,
          "quantity": 2
        }
      ]
    }
  },
  {
    "purchase": {
      "customer": {
        "loyaltyId": 5512,
        "firstName": "Jane",
        "lastName": "Doe"
      },
      "total": 121.44,
      "items": [
        {
          "itemDescription": "book",
          "price": 28.15,
          "quantity": 4
        },
        {
          "itemDescription": "inkpen",
          "price": 4.42,
          "quantity": 2
        }
      ]
    }
  }
]

与以下富化内容合并：

[
  {
    "customerDetails": {
      "id": 48202,
      "phone": "555-555-5555",
      "email": "john.doe@lynxus.acal.org.cn"
    }
  },
  {
    "customerDetails": {
      "id": 5512,
      "phone": "555-555-5511",
      "email": "jane.doe@lynxus.acal.org.cn"
    }
  }
]

再假设使用的 Record Path 值为 "/purchase/customer"，会得到如下结果：

[
  {
    "purchase": {
      "customer": {
        "loyaltyId": 48202,
        "firstName": "John",
        "lastName": "Doe",
        "customerDetails": {
          "id": 48202,
          "phone": "555-555-5555",
          "email": "john.doe@lynxus.acal.org.cn"
        }
      },
      "total": 48.28,
      "items": [
        {
          "itemDescription": "book",
          "price": 24.14,
          "quantity": 2
        }
      ]
    }
  },
  {
    "purchase": {
      "customer": {
        "loyaltyId": 5512,
        "firstName": "Jane",
        "lastName": "Doe",
        "customerDetails": {
          "id": 5512,
          "phone": "555-555-5511",
          "email": "jane.doe@lynxus.acal.org.cn"
        }
      },
      "total": 121.44,
      "items": [
        {
          "itemDescription": "book",
          "price": 28.15,
          "quantity": 4
        },
        {
          "itemDescription": "inkpen",
          "price": 4.42,
          "quantity": 2
        }
      ]
    }
  }
]

SQL

SQL 策略提供了区别于其他策略的重要能力：它允许以索引之外的方式关联 "original" FlowFile 与 "enrichment" FlowFile 中的记录。也就是说， 基于 SQL 的策略不一定要把原始 FlowFile 的第一条记录与富化 FlowFile 的第一条记录配对，而是允许使用标准的 SQL JOIN 表达式来关联记录。

一个常见用例是构造负载去查询某个 Web 服务。响应包含用于富化的标识符及 附加信息，但富化记录的顺序可能与原始记录的顺序不一致。

举例来说，考虑以下 CSV 格式的原始负载：

id, name, age
28021, John Doe, 55
832, Jane Doe, 22
29201, Jake Doe, 23
555, Joseph Doe, 2

再考虑以下富化数据负载：

customer_id, customer_email, customer_name, customer_since
555, joseph.doe@lynxus.acal.org.cn, Joe Doe, 08/Dec/14
832, jane.doe@lynxus.acal.org.cn, Mrs. Doe, 14/Nov/14
28021, john.doe@lynxus.acal.org.cn, John Doe, 22/Jan/22

使用 SQL 策略时，必须提供一条 SQL SELECT 语句，把原始数据和富化数据合并 到单个 FlowFile 中。做法是把原始 FlowFile 视为名为 "original" 的表， 把富化数据视为名为 "enrichment" 的表。

基于此，可以用一条简单的查询合并所有数据：

SELECT o.*, e.*
FROM original o
JOIN enrichment e
ON o.id = e.customer_id

输出如下：

id, name, age, customer_id, customer_email, customer_name, customer_since
28021, John Doe, 55, 28021, john.doe@lynxus.acal.org.cn, John Doe, 22/Jan/22
832, Jane Doe, 22, 832, jane.doe@lynxus.acal.org.cn, Mrs. Doe, 14/Nov/14
555, Joseph Doe, 2, 555, joseph.doe@lynxus.acal.org.cn, Joe Doe, 08/Dec/14

注意，本例中 Jake Doe 的记录被移除了，因为使用的是 JOIN 而不是 OUTER JOIN。 可以改用 LEFT OUTER JOIN，以保留原始 FlowFile 的所有记录，并为富化中缺失 的记录提供 null 值：

SELECT o.*, e.*
FROM original o
LEFT OUTER JOIN enrichment e
ON o.id = e.customer_id

输出如下：

id, name, age, customer_id, customer_email, customer_name, customer_since
28021, John Doe, 55, 28021, john.doe@lynxus.acal.org.cn, John Doe, 22/Jan/22
832, Jane Doe, 22, 832, jane.doe@lynxus.acal.org.cn, Mrs. Doe, 14/Nov/14
29201, Jake Doe, 23,,,,
555, Joseph Doe, 2, 555, joseph.doe@lynxus.acal.org.cn, Joe Doe, 08/Dec/14

但 SQL 的表达能力远不止于此，可以执行强大得多的表达式。本例中我们可能 并不想同时保留 "id" 和 "customer_id" 字段，或 "name" 和 "customer_name" 字段。假设富化提供的是客户的偏好名（preferred name）而非法定名，我们可能 想去掉 customer_since 列（它对用例没有意义）。于是可以把 SQL 改为：

SELECT o.id, o.name, e.customer_name AS preferred_name, o.age, e.customer_email AS email
FROM original o
         LEFT OUTER JOIN enrichment e
                         ON o.id = e.customer_id

这会产生更方便的输出：

id, name, preferred_name, age, email
28021, John Doe, John Doe, 55, john.doe@lynxus.acal.org.cn
832, Jane Doe, Mrs. Doe, 22, jane.doe@lynxus.acal.org.cn
29201, Jake Doe,, 23,
555, Joseph Doe, Joe Doe, 2, joseph.doe@lynxus.acal.org.cn

由此可见 SQL 策略的强大。不过，使用 SQL 策略时必须考虑一个非常重要的问题。

警告： SQL 策略虽然强大，但可能需要大量堆内存。取决于查询本身，SQL 引擎可能需要把整个 "enrichment" FlowFile 的内容缓冲到 Java 堆内存中。 此外，如果 Processor 配置了多个并发任务，每个任务都可能把整个富化 FlowFile 的内容保存在内存中。这可能导致堆耗尽，引发稳定性问题或 OutOfMemoryError。

有几个选项可以帮助缓解这些问题：

1. 拆分为较小的块。通常不建议把面向 Record 的数据拆成大量小 FlowFile，因为 Lynxus 在较大的 FlowFile 上表现最好，最佳区间大约在 300 KB 到 3 MB 之间。因此，我们不想把一个含 100,000 条记录的大 FlowFile 拆成 100,000 个各含 1 条记录的 FlowFile。但在 ForkEnrichment Processor 之前，把它拆成 100 个各含 1,000 条记录的 FlowFile，或 10 个各含 10,000 条记录的 FlowFile，可能是有利的。 这通常会使富化数据量变小，从而不必在内存中保存那么多内容。
2. 在 JoinEnrichment Processor 之前，裁剪富化数据，移除不需要的字段。 在上面的例子中，我们可以使用 QueryRecord、UpdateRecord、 JoltTransformRecord，或更新 schema，从富化数据集中移除 "customer_since" 字段。由于不使用该字段，可以在 JoinEnrichment 之前轻松移除它，以减小富化 FlowFile 的大小，从而减少内存中保存的 数据量。

还值得注意的是，SQL 策略可能导致 FlowFile 内记录的顺序发生变化，因此 如果顺序很重要，可能需要使用 ORDER BY 子句等。

其他内存注意事项

除了上面关于使用 SQL Join 策略的警告之外，为了限制该 Processor 必须保存 在内存中的信息量，还有另一个需要考虑的问题。虽然该 Processor 不会把 所有 FlowFile 的内容都保存在内存中，但它会把所有 FlowFile 的 Attribute 保存在内存中。因此，使用该 Processor 时应注意以下几点：

1. 避免大 Attribute。FlowFile Attribute 不应用于保存 FlowFile 内容。 Attribute 应当很小，通常在 100-200 个字符量级。如果存在大 Attribute，建议在 ForkEnrichment Processor 之前用 UpdateAttribute Processor 移除它们。
2. 避免大量 Attribute。避免创建大的 FlowFile Attribute 很重要，避免 创建大量 Attribute 同样重要。一个 FlowFile 上保留 30 个小 Attribute 完全没问题，但存储 300 个 Attribute 可能占用大量堆。
3. 限制背压。JoinEnrichment Processor 会把所有传入的 FlowFile 拉入自身内存。因此，避免在任何时候向 Processor 提供过多 FlowFile 会有帮助。做法是把背压限制设为较小的值。例如在上面的 例子中，ForkEnrichment Processor 直接连接到 JoinEnrichment Processor，可以把该连接的背压限制设为 500 或 1,000 而不是默认的 10,000。这可以限制一次加载到 JoinEnrichment Processor 中的 FlowFile 数量。

更复杂的合并策略

该 Processor 提供了若干策略，用于关联数据并把两个不同 FlowFile 的记录 合并到一个 FlowFile 中。但有时用户可能需要比这些更强大的能力。例如， 我们可能想利用富化记录中的信息来决定是否把对应原始记录中的某个值 置为 null。

对于这类用例，推荐的做法是使用 Wrapper 策略或 SQL 策略把原始 FlowFile 和富化 FlowFile 合并为单个 FlowFile，然后把这个 Processor 的 "joined" Relationship 连接到最适合进一步处理数据的 Processor。 例如，假设使用 Wrapper 策略产生如下输出：

{
  "original": {
    "id": 482028,
    "name": "John Doe",
    "ssn": "555-55-5555",
    "phone": "555-555-5555",
    "email": "john.doe@lynxus.acal.org.cn"
  },
  "enrichment": {
    "country": "UK",
    "allowsPII": false
  }
}

接着可以使用 TransformRecord Processor，配合 JSON RecordReader 和 JSON RecordSetWriter 进行转换。使用 Groovy 时，转换逻辑可能如下：

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

Record original = (Record) record.getValue("original")
Record enrichment = (Record) record.getValue("enrichment")

if (Boolean.TRUE != enrichment?.getAsBoolean("allowsPII")) {
    original.setValue("ssn", null)
    original.setValue("phone", null)
    original.setValue("email", null)
}

return original

这会产生如下输出：

{
  "id": 482028,
  "name": "John Doe",
  "ssn": null,
  "phone": null,
  "email": null
}

通过这种方式，我们利用富化记录中的信息选择性地转换了原始记录，然后返回 原始记录，完全丢弃富化记录。这为基于所获取的富化数据内容来转换原始负载 提供了无限的可能性。