JsonPathReader

JsonPathReader Controller Service 解析 JSON 格式的 FlowFile。用户自定义属性 指定如何从 JSON 中提取所有相关字段以创建记录。除非至少提供一个 JSON Path， 否则该 Controller Service 无效。与 JsonTreeReader Controller Service 不同， 该服务返回的记录只包含通过 JSON Path 配置的字段。

如果 FlowFile 的 JSON 根节点是 JSON 数组，数组中的每个 JSON 对象会被视为 一条独立的记录，而不是一条由数组组成的记录。如果根节点是 JSON 对象， 则作为单条记录求值。

提供 JSON Path 的方式是添加用户自定义属性：属性名即返回记录中对应字段的 名称，属性值必须是合法的 JSON Path 表达式。该 JSON Path 会对 FlowFile 中每个 顶层 JSON 对象求值，求值结果就是属性名所指定字段的值。如果给定了 JSON Path 但 schema 中没有对应名称的字段，该字段会被跳过。

该 Controller Service 必须配置 schema。每个被求值且出现在 schema"根层级" 的 JSON Path 都会在记录中产生一个字段。也就是说，schema 应当与求值所有 JSON Path 后创建的记录相匹配，而不是与从 FlowFile 读入的"传入 JSON"相匹配。

Schema 与类型强制转换

从传入数据解析记录时，数据会被拆分为各个字段。每个字段随后会按字段名与 配置的 schema 进行比对，以确定数据的类型。如果字段不在 schema 中，该字段 会从记录中省略。如果在 schema 中找到了该字段，接收到的数据的数据类型会 与 schema 指定的数据类型比较。如果类型一致，字段值原样使用。如果 schema 表明该字段应为其他类型，Controller Service 会尝试把数据强制转换为 schema 指定的类型。如果无法强制转换，会抛出异常。

尝试把一个字段值从一种数据类型强制转换为另一种时，适用以下规则：

• 任何数据类型都可以强制转换为 String 类型。
• 任何数值类型（Byte、Short、Int、Long、Float、Double）都可以强制转换为其他数值类型。
• 任何数值都可以强制转换为 Date、Time 或 Timestamp 类型，方式是把 Long 值 视为自纪元（1970 年 1 月 1 日午夜，GMT）以来的毫秒数。
• 如果 String 值的格式与配置的 "Date Format"、"Time Format" 或 "Timestamp Format" 匹配，就可以强制转换为 Date、Time 或 Timestamp 类型。
• 如果值的类型合适，String 值可以强制转换为数值。例如，String 值 8 可以强制转换为任何数值类型；但 String 值 8.2 只能强制转换为 Double 或 Float 类型，不能转为 Integer。
• String 值 "true" 或 "false"（不区分大小写）可以强制转换为 Boolean 值。
• 非空的 String 值可以强制转换为 Char 类型。如果字符串包含多个字符， 只使用第一个字符，其余忽略。
• 任何"日期/时间"类型（Date、Time、Timestamp）都可以强制转换为其他"日期/时间"类型。
• 任何"日期/时间"类型都可以强制转换为 Long 类型，表示自纪元（1970 年 1 月 1 日午夜，GMT）以来的毫秒数。
• 任何"日期/时间"类型都可以强制转换为 String。String 的格式就是对应属性 （Date Format、Time Format、Timestamp Format）所配置的 DateFormat。

如果尝试在两种数据类型之间强制转换时上述规则都不适用，转换会失败并抛出异常。

Schema 推断

虽然 Lynxus 的 Record API 要求每条记录都有 schema，但基于数据中的值来推断 schema 往往比手动创建 schema 更方便。做法是把 "Schema Access Strategy" 属性设为 "Infer Schema"。使用该策略时，Reader 会先解析 FlowFile 中的所有 数据，记录遇到的所有字段及每个字段的类型。所有数据解析完成后，会形成 一个涵盖所有已遇到字段的 schema。

推断 schema 时的一个常见顾虑是如何处理两个值类型不同的情况。例如，考虑 包含以下两条记录的 FlowFile：

[
  {
    "name": "John",
    "age": 8,
    "values": "N/A"
  },
  {
    "name": "Jane",
    "age": "Ten",
    "values": [
      8,
      "Ten"
    ]
  }
]

显然 "name" 字段会被推断为 STRING 类型。但 "age" 字段该如何处理？应该是 INT 和 STRING 之间的 CHOICE 吗？应该优先用 LONG 而不是 INT 吗？直接用 STRING 吗？该字段应视为可为 null 吗？

为帮助理解该 Record Reader 如何推断 schema，以下是推断逻辑所遵循的规则：

• 所有字段都被推断为可为 null。
• 当两条不同记录中同一字段遇到两个值（或 ARRAY 类型遇到两个值）时， 推断引擎倾向于使用"更宽"的数据类型，而不是 CHOICE 类型。当且仅当数据 类型 "A" 除了包含 "B" 的所有值外还包含其他值时，才称数据类型 "A" 比 数据类型 "B" 宽。例如，LONG 类型比 INT 宽，但不比 BOOLEAN 宽 （BOOLEAN 也不比 LONG 宽）。INT 比 SHORT 宽。STRING 类型被认为比除 MAP、RECORD、ARRAY、CHOICE 之外的所有类型都宽。
• 当两条不同记录中同一字段遇到两个值（或 ARRAY 类型遇到两个值）， 但两个值的类型互不比对方宽时，使用 CHOICE 类型。在上面的例子中， "values" 字段会被推断为 STRING 或 ARRAY 之间的 CHOICE。
• 如果配置了 "Time Format"、"Timestamp Format" 或 "Date Format" 属性， 任何原本会被视为 STRING 类型的值，都会先与配置的格式比对，看是否匹配。 如果值匹配 Timestamp Format，视为 Timestamp 字段；匹配 Date Format 视为 Date 字段；匹配 Time Format 视为 Time 字段。如果值碰巧匹配多个 配置的格式，按 Timestamp、Date、Time 的顺序匹配。也就是说，如果一个值 同时匹配 Timestamp Format 和 Date Format，推断出的类型是 Timestamp。 由于解析日期和时间开销较大，如果不预期出现日期、时间和时间戳，或者可以 接受按 STRING 处理数据，建议不要配置这些格式。对于需要此功能的用例， 推断引擎足够智能，会先检查若干非常廉价的条件来优化解析。例如，先检查 字符串长度是否过长或过短、不可能匹配模式。这比尝试把每个字符串值都 解析为时间戳要高效得多。
• MAP 类型永远不会被推断，而是使用 RECORD 类型。
• 如果字段存在但所有值都是 null，该字段被推断为 STRING 类型。

推断 Schema 的缓存

该 Record Reader 要求：如果要推断 schema，必须读取所有记录，以确保推断出的 schema 适用于 FlowFile 中的所有记录。但这可能开销很大，尤其是数据要经过 很多不同转换时。为降低推断成本，可以通过设置同名属性为该 Record Reader 配置 "Schema Inference Cache"。这是一个可以在 Record Reader 和 Record Writer 之间共享的 Controller Service。

每当使用 Record Writer 写出数据时，如果它配置了 "Schema Cache"，也会把 schema 添加到 Schema Cache 中。这会使该 schema 的标识符作为 Attribute 添加到 FlowFile 上。

每当使用 Record Reader 读取数据时，如果它配置了 "Schema Inference Cache"， 会先在 FlowFile 上查找 "schema.cache.identifier" Attribute。如果该 Attribute 存在，就用它的值在 schema 缓存中查找 schema。如果能在缓存中 找到该标识符对应的 schema，就直接使用，而不必读取、解析和分析数据来 推断。如果 FlowFile 上没有该 Attribute，或缓存中没有对应标识符的 schema，Record Reader 会按上述方式推断。

最终效果是：用户可以串联许多不同的 Processor 来处理面向 Record 的数据。 通常只有链中的第一个 Processor 需要承担推断 schema 的"代价"。对链中的 其他所有 Processor，Record Reader 可以直接按标识符在 Schema Cache 中 查找。这使得 Record Reader 既能准确推断（基于 FlowFile 中的所有数据）， 又能高效执行（无论多少 Processor 处理数据，通常只推断一次）。

示例

举例来说，考虑一个内容包含以下 JSON 的 FlowFile：

[
  {
    "id": 17,
    "name": "John",
    "child": {
      "id": "1"
    },
    "siblingIds": [
      4,
      8
    ],
    "siblings": [
      {
        "name": "Jeremy",
        "id": 4
      },
      {
        "name": "Julia",
        "id": 8
      }
    ]
  },
  {
    "id": 98,
    "name": "Jane",
    "child": {
      "id": 2
    },
    "gender": "F",
    "siblingIds": [],
    "siblings": []
  }
]

并配置了以下 schema：

{
  "namespace": "lynxus",
  "name": "person",
  "type": "record",
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "childId",
      "type": "long"
    },
    {
      "name": "gender",
      "type": "string"
    },
    {
      "name": "siblingNames",
      "type": {
        "type": "array",
        "items": "string"
      }
    }
  ]
}

如果为该 Controller Service 配置以下用户自定义属性：

此时，该 FlowFile 会生成两条记录。第一条记录包含以下键/值对：

第二条记录包含以下键/值对：