CEFReader

CEFReader Controller Service 提供了一种读取和解析 CEF 消息的手段。

该 Reader 支持 CEF Version 23。预期格式以及 Extension Dictionary 所知的扩展字段 由 ArcSight Common Event Format 的描述定义。该 Reader 支持处理 Syslog 前缀和 自定义扩展。以下是 Reader 可以处理的一些 CEF 消息示例：

CEF:0|Company|Product|1.2.3|audit-login|Successful login|3| 
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3| 
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23  dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789 loginsequence=123

原始消息（Raw message）

可以在生成的记录中保留原始消息。当消息包含不属于记录实例的 Syslog 前缀时， 这一点很有用。要保留原始消息，必须设置 "Raw Message Field" 属性。Reader 会把 该属性的值用作字段名，并把原始消息作为自定义扩展字段添加。"Raw Message Field" 的值必须与 CEF Extension Dictionary 所知的头字段（header fields）和扩展字段 不同。如果该属性为空，则不会添加原始消息。

使用预定义 schema 时，"Raw Message Field" 定义的字段必须以 STRING 记录字段 的形式出现在 schema 中。如果 schema 是推断出来的，无论推断策略是什么，该字段 都会作为额外的自定义扩展自动添加。

Schema 与类型强制转换（Type Coercion）

从传入数据解析记录时，数据会被拆分为各个字段。每个字段随后会按字段名与配置的 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 值的第一个 字符是数字，可以使用该字符，其余字符会被忽略。
• 任何"日期/时间"类型（Date、Time、Timestamp）都可以强制转换为其他"日期/时间"类型。
• 任何"日期/时间"类型都可以强制转换为 Long 类型，表示自纪元（1970 年 1 月 1 日 午夜，GMT）以来的毫秒数。
• 任何"日期/时间"类型都可以强制转换为 String。String 的格式就是对应属性 （Date Format、Time Format、Timestamp Format 属性）所配置的 DateFormat。

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

Schema 推断（Schema inference）

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

推断 schema 时的一个常见顾虑是如何处理两个值类型不同的情况。例如，某个自定义 扩展字段在一条记录中是 Float 值，在另一条记录中是 String。这种情况下，推断 出的结果会包含一个具有 FLOAT 和 STRING 选项的 CHOICE 数据类型，记录在该特定 字段上允许使用任一值。

CEF 格式不仅规定了消息格式，还规定了内容的类型。因此，某些字段的数据类型不由 FlowFile 中的实际值决定，而是由 CEF 格式决定。这包括头字段——它们必须始终 出现，并符合 CEF 格式定义的数据类型。此外，Extension Dictionary 中的扩展字段 是否出现在生成的 schema 中，取决于 FlowFile 的内容；但一旦添加了扩展字段， 其数据类型受 CEF 格式约束。自定义扩展没有类似限制，它们是否出现在 schema 中 完全取决于 FlowFile 的内容。

为方便起见，CEFReader 的 schema 推断支持多种策略。这些策略决定应把传入 CEF 消息中的哪些字段纳入 schema。例如，可以用它过滤掉自定义扩展。需要注意 的是，这会对记录处理的后续步骤产生重大影响。例如，使用会省略字段的 Inference Strategy 并配合 ConvertRecord Processor，会导致记录只保留原始 字段的一部分。

Headers only（仅头字段）

使用该策略得到的 schema 只包含传入消息中的头字段。所有其他字段（标准扩展或 自定义扩展）都会被忽略。这些字段的类型由 CEF 格式定义，无论用作模板的消息 内容是什么，其数据类型也由格式定义。

Headers and extensions（头字段加扩展字段）

除头字段外，该策略还会包含 FlowFile 中消息里的标准扩展。也就是说，输出 Record 中不会包含所有标准扩展，只包含 Schema Inference 在传入消息中找到的 那些。这些记录字段的数据类型由 CEF 格式决定，忽略所观察字段中的实际值。

With custom extensions inferred（推断自定义扩展）

头字段和标准扩展字段的类型受 CEF 格式约束，而消息中还可以添加称为 "自定义扩展"的额外字段。这些字段不属于 "Extension Dictionary"，因此没有 预定义的数据类型。使用 "With custom extensions inferred" 推断策略时， CEFReader 会尝试根据值确定这些自定义扩展字段的可能数据类型。

With custom extensions as strings（自定义扩展作为字符串）

某些情况下不希望由 Reader 决定自定义扩展的类型。为方便起见，CEFReader 提供了 一种推断策略，无论值是什么，都把自定义扩展字段视为 String 数据。除此之外， 该策略的行为与 "With custom extensions inferred" 相同。

推断 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，就直接使用它，而不必读取、解析和分析数据来推断 schema。如果 FlowFile 上没有该 Attribute，或该 Attribute 存在但缓存中没有 对应标识符的 schema，Record Reader 会按上述方式推断 schema。

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

处理非法事件

如果事件的格式有问题，导致底层 CEF 解析器无法正确读取，该事件被视为非法。 CEFReader 有两种处理非法事件的方式，由 "Invalid Field" 属性决定。如果该 属性未设置，读取会在遇到第一个非法事件时失败。如果该属性已设置，读取产物 会是一条只有一个字段的记录：字段名基于该属性，字段值就是原始事件文本。 默认情况下，"Invalid Field" 属性未设置。

当 "Invalid Field" 属性已设置时，读取出的记录可能同时包含格式正确的 CEF 事件记录和格式错误的记录。因此，在进一步处理之前，可能需要额外步骤 把它们分开。