mysql - QueryRecord 处理器在 NiFi 中执行聚合 SQL 函数

Question

我有以下流程:

QueryDatabaseTable ->QueryRecord -> UpdateAttribute->MergeContent->PutelasticsearchHttp

想法是从数据库中获取记录并对字段执行聚合函数。在我的数据库表中，我有以下 4 个字段:

DeptId  DepartmentName  Address  ExperienNo

1    DS    San Jose      4

2    GT    San Fran      6

3    At    Oakland       8

4    BMS   detroit       3

5    RT    Haawai        9

并且我已将 QueryREcord 的 Controller 服务设置为 Avroreader 和 CSVSetWritter，其架构如下:

{
"type": "record",
"name": "SQLSchema",
"fields" : [
{"name": "DeptId", "type": "int"},
{"name": "DepartmentName", "type": "string"},
{"name": "Address", "type": "string"},
{"name": "ExperienceNo", "type": "int"},
{"name": "Total_Experience", "type": "int"}
]
}

我想对两个字段(DeptId-ExperienceNo)进行差分

我的 SQL 查询如下:

SELECT DeptId, DepartmentName,Address,ExperienceNo,
(DeptId - ExperienceNo) AS Total_Experience FROM flowfile

我收到错误，因为“Total_Experience”不能为空。

然而，同样的查询在 MySQL 中运行良好。这是如何实现的，我可以在字段上执行 SQL 聚合函数并将其别名为新的动态列。

谢谢。

非常感谢任何建议。

Answer 1

您的csv 阅读器 Controller 服务配置为

{"name": "Total_Experience", "type": "int"} 字段没有默认值，在你的输入数据中你没有得到这个字段.

因此处理器正在提示 Total_Experience 不能为null。

要解决此问题，请更改 avro 架构 以包含 Total_Experience 字段的 null 值。

Avro Schema:

{
"type": "record",
"name": "SQLSchema",
"fields" : [
{"name": "DeptId", "type": "int"},
{"name": "DepartmentName", "type": "string"},
{"name": "Address", "type": "string"},
{"name": "ExperienceNo", "type": "int"},
{"name": "Total_Experience", "type": ["null","int"]}
]
}

CsvReader Controller 服务配置:

输出:

DeptId,DepartmentName,Address,ExperienceNo,Total_Experience
1,DS,San Jose,4,-3
2,GT,San Fran,6,-4
3,At,Oakland,8,-5
4,BMS,detroit,3,1
5,RT,Haawai,9,-4

正确的做法是:

在没有 Total_Experience 字段的情况下配置 CSV 阅读器 和在 CsvSetWriter Controller 服务中包含 Total_Experience，因为您通过使用查询记录 处理器创建此字段。