QueryRecord

对流数据执行 SQL

QueryRecord 借助一种广为人知的语法（SQL），在数据流经系统时对其进行路由、过滤、转换和查询， 为用户提供了强大的能力。为了最大程度地保持灵活性，该 Processor 配置了一个负责读取和解析 传入 FlowFile 的 Controller Service，以及一个负责写出结果的 Controller Service。 采用这种范式，用户不必为了查询而把数据从一种格式转换成另一种格式、查询完再转回想要的格式； 只需为相应的数据格式配置合适的 Controller Service 即可使用。

该 Processor 不提供单一的 "SQL SELECT Statement" 式属性，而是使用用户自定义属性。 添加到 Processor 的每个用户自定义属性，其名称会成为 Processor 的一个新 Relationship， 并对应一条将对每个 FlowFile 求值的 SQL 查询。这样，每个 FlowFile 上可以运行多条 SQL 查询。

该 Processor 支持的 SQL 语法是 ANSI SQL，由 Apache Calcite 驱动。请注意，标识符使用 双引号包裹，列名/别名不区分大小写。

举例来说，假设我们有一个包含以下 CSV 数据的 FlowFile：

name, age, title
John Doe, 34, Software Engineer
Jane Doe, 30, Program Manager
Jacob Doe, 45, Vice President
Janice Doe, 46, Vice President

再假设我们给 Processor 添加了以下属性：

此时该 Processor 会有五个 Relationship：original、failure、Engineers、VP 和 Younger Than Average。如果处理 FlowFile 时失败，原始 FlowFile 会被路由到 failure。 否则，原始 FlowFile 会被路由到 original，并且每个其他 Relationship 上各路由一个 FlowFile，取值如下：

注意，此示例旨在说明 Processor 输入和输出的数据。数据的实际格式可能因所使用的 Record Reader 和 Record Writer 的配置而异。例如，这里假设使用带表头行的 CSV Reader 和 CSV Writer。如果改用 JSON Writer，输出会包含相同的信息，但以 JSON 形式呈现。 用户可以自行选择最适合其用例的输入和输出格式，输入和输出格式不必相同。

还值得注意的是，输出 FlowFile 有两种不同的 schema。Engineers 和 Younger Than Average 的 FlowFile 包含 3 个字段：name、age 和 title，而 VP 的 FlowFile 只包含 name 字段。在大多数情况下，Record Writer 被配置为使用 Record 所提供的 Schema（通常意味着其 Schema Access Strategy 配置为 Inherit Record Schema）， 这样运行良好。但如果显式地为 Record Writer 提供了 Schema，就必须确保该 Schema 覆盖了所有字段；否则，Record Writer 的 schema 中缺少的字段将不会出现在输出中。

对层次化数据执行 SQL

在对任意数据流执行 SQL 时，必须考虑的一个重要细节是：如何处理层次化数据，例如 JSON、XML 和 Avro。由于 SQL 最初是为关系型数据库开发的，而关系型数据库表示的是 "扁平"数据，因此很容易理解它如何映射到 CSV 文件等其他"扁平"数据，甚至所有字段都是 原始类型的"扁平" JSON。但在很多情况下，用户希望对由许多嵌套值组成的 JSON 或 Avro 数据执行 SQL。例如，考虑以下 JSON 作为输入：

            {
  "name": "John Doe",
  "title": "Software Engineer",
  "age": 40,
  "addresses": [
    {
      "streetNumber": 4820,
      "street": "My Street",
      "apartment": null,
      "city": "New York",
      "state": "NY",
      "country": "USA",
      "label": "work"
    },
    {
      "streetNumber": 327,
      "street": "Small Street",
      "apartment": 309,
      "city": "Los Angeles",
      "state": "CA",
      "country": "USA",
      "label": "home"
    }
  ],
  "project": {
    "name": "ACAL Lynxus",
    "maintainer": {
      "id": 28302873,
      "name": "Apache Software Foundation"
    },
    "debutYear": 2014
  }
}

考虑一个查询：选出家庭住址所在州与工作地点所在州不同的人的 title 和 name。这里我们 只能选择 name、title、age 和 addresses 字段。在此场景中，addresses 表示一个由复杂对象（即记录）组成的数组。为了适应这种情况，QueryRecord 提供了 用户自定义函数以启用 Record Path。Record Path 是一种简单的 Lynxus 领域专用 语言（DSL），允许用户引用嵌套结构。

主要使用的用户自定义函数名为 RPATH（Record Path 的缩写）。该函数恰好接受两个参数： 要对其求值 RecordPath 的 Record，以及要执行的 RecordPath（按此顺序）。因此， 要选出家庭住址所在州与工作地点所在州不同的人的 title 和 name，可以使用以下 SQL 语句：

SELECT title, name
FROM FLOWFILE
WHERE RPATH(addresses, '/state[/label = ''home'']') <> RPATH(addresses, '/state[/label = ''work'']')

用自然语言解释这条查询：它从 FlowFile 中选出满足以下条件的记录的 "title" 和 "name" 字段：存在一个 "label" 为 "home" 的地址和另一个 "label" 为 "work" 的地址， 且这两个地址的州不同。

类似地，我们可以用以下查询选出 "project" 的维护者是 Apache Software Foundation 的 人的完整记录（所有字段）：

SELECT *
FROM FLOWFILE
WHERE RPATH(project, '/maintainer/name') = 'Apache Software Foundation'

不过，使用 RecordPath 有一个注意事项：RPATH 函数返回的是一个 Object，在 JDBC 中表示为 OTHER 类型。像上面那样使用时没有问题，不影响任何行为。但如果我们想对 结果再使用另一个 SQL 函数呢？例如，想使用 SELECT * FROM FLOWFILE WHERE RPATH(project, '/maintainer/name') LIKE 'Apache%' 这样的查询？这将会失败，并产生很长的错误，例如：

3860 [pool-2-thread-1] ERROR cn.org.acal.lynxus.processors.standard.QueryRecord - QueryRecord[id=135e9bc8-0372-4c1e-9c82-9d9a5bfe1261] Unable to query FlowFile[0,174730597574853.mockFlowFile,0B] due to java.lang.RuntimeException: Error while compiling generated Java code: ... Caused by: org.codehaus.commons.compiler.CompileException: Line 21, Column 180: No applicable constructor/method found for actual parameters "java.lang.Object, java.lang.String"; candidates are: "public static boolean org.apache.calcite.runtime.SqlFunctions.like(java.lang.String, java.lang.String)", "public static boolean org.apache.calcite.runtime.SqlFunctions.like(java.lang.String, java.lang.String, java.lang.String)"

这是因为 LIKE 函数期望用它来比较 String 对象，即期望的格式是 String LIKE String， 而我们传入的是 Other LIKE String。为此，还提供了几个其他 RecordPath 函数： RPATH_STRING、RPATH_INT、RPATH_LONG、RPATH_FLOAT 和 RPATH_DOUBLE， 分别用于让返回类型成为 String、Integer、Long（64 位整数）、Float 或 Double。因此，上面的查询需要改写为 SELECT * FROM FLOWFILE WHERE RPATH_STRING(project, '/maintainer/name') LIKE 'Apache%'， 这样就能产生期望的输出。

聚合函数

为了对数据流执行 SQL，该 Processor 把每个单独的 FlowFile 视为一张独立的表。 因此，SUM 和 AVG 等聚合函数会对每个 FlowFile 内的所有记录求值，但不会跨越 FlowFile 边界。举例来说，考虑一个包含以下数据的 CSV 格式输入 FlowFile：

name, age, gender
John Doe, 40, Male
Jane Doe, 39, Female
Jimmy Doe, 4, Male
June Doe, 1, Female

对于这些数据，我们可能想执行带聚合函数（例如 MAX）的查询：

SELECT name
FROM FLOWFILE
WHERE age = (SELECT MAX(age))

上面的查询会选出年龄最大的人的名字，即 John Doe。如果随后到达第二个 FlowFile，其内容会作为一张全新的表来求值。