CSVReader

CSVReader 可以把输入数据解析为带分隔符的记录（Record）。默认使用逗号作为 字段分隔符，但这是可配置的。例如，使用制表符来读取以制表符分隔的值（TSV） 是一种常见做法。 Reader 内置了预定义的 CSV 格式（如 EXCEL）。关于这些格式配置的更多信息 参见：https://commons.apache.org/proper/commons-csv/apidocs/org/apache/commons/csv/CSVFormat.html Reader 允许自定义 CSV 格式，例如用什么字符分隔 CSV 字段、用什么字符做 引号以及何时对字段加引号、用什么字符表示注释等。字段名的指定方式有两种： 以"表头行（header line）"作为 CSV 的第一行（此时 Schema Access Strategy 应为 "Infer Schema" 或 "Use String Fields From Header"），或者通过 Schema Text 指定、或在 Schema Registry 中查找 schema 来提供。

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 Access Strategy" 属性设为 "Infer Schema"。使用该策略时，Reader 会先解析 FlowFile 中的所有数据， 记录遇到的所有字段及每个字段的类型。所有数据解析完成后，会形成一个涵盖 所有已遇到字段的 schema。

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

name, age
John, 8
Jane, 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 之外的所有类型都宽。
• 在推断值的类型之前，会先去掉首尾空白。此外，如果值被双引号（"）包围， 双引号会被去掉。因此，值 16 与 "16" 的解释方式相同，都会被解释为 INT。但值 " 16" 会被推断为 STRING 类型，因为空白被包在双引号内， 意味着空白被视为值的一部分。
• 如果配置了 "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 类型永远不会被推断。
• ARRAY 类型永远不会被推断。
• 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 处理数据，通常只推断一次）。

示例

示例 1

举例来说，考虑一个内容如下、带注释的 FlowFile：

id, name, balance, join_date, notes
1, John, 48.23, 04/03/2007, "Our very   first customer!"
2, Jane, 1245.89, 08/22/2009,	  
3, Frank Franklin, "48481.29", 04/04/2016,  

再假设该 Controller Service 配置的 Schema Registry 指向 AvroSchemaRegistry，schema 配置如下：

{
  "namespace": "lynxus",
  "name": "balances",
  "type": "record",
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "balance",
      "type": "double"
    },
    {
      "name": "join_date",
      "type": {
        "type": "int",
        "logicalType": "date"
      }
    },
    {
      "name": "notes",
      "type": "string"
    }
  ]
}

在上面的例子中，'join_date' 列是 Date 类型。为了让 CSV Reader 能够正确 地把值解析为日期，需要为 Reader 提供要使用的日期格式。本例中，我们把 Date Format 属性配置为 MM/dd/yyyy，表示两位月份、两位日期、四位年份， 以斜杠分隔。此时，该 FlowFile 会生成 3 条不同的记录。第一条记录包含 以下值：

第二条记录包含以下值：

第三条记录包含以下值：

示例 2 - 带 CSV 表头行的 Schema

当 CSV 数据包含描述列名的表头行时，Reader 提供了若干属性来配置如何处理 这些列名。"Schema Access Strategy" 属性及相关属性（"Schema Registry"、 "Schema Text" 和 "Schema Name"）可用于指定 schema 的获取方式。如果 "Schema Access Strategy" 设为 "Use String Fields From Header"，CSV 的 表头行会被用来确定 schema；否则 schema 从其他地方引用。但如果从 Schema Registry 获取的 schema 与 CSV 表头指示的列名不一致，会发生什么？

例如，假设从 Schema Registry 获取了以下 schema：

{
  "namespace": "lynxus",
  "name": "balances",
  "type": "record",
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "balance",
      "type": "double"
    },
    {
      "name": "memo",
      "type": "string"
    }
  ]
}

而 CSV 包含以下数据：

id, name, balance, notes 1, John Doe, 123.45, First Customer

注意，schema 表明最后一列名为 "memo"，而 CSV 表头指示的名称是 "notes"。

此时，Reader 会查看 "Ignore CSV Header Column Names" 属性。如果该属性 设为 "true"，CSV 中提供的列名会被直接忽略，最后一列将名为 "memo"。 但如果 "Ignore CSV Header Column Names" 属性设为 "false"，结果是最后 一列名为 "notes"，每条记录的 "memo" 列都是 null 值。

"Ignore CSV Header Column Names" 属性设为 "false" 时：

"Ignore CSV Header Column Names" 属性设为 "true" 时：