将序列号添加到apache nifi中的csv中的列

Question

我正在 Apache Nifi 中读取 CSV 文件，现在我想添加一个序列号列，该列将包含 CSV 文件中所有行的序列号。因此，如果我有 10 行，则序列号将从 1 到 10。我如何通过 Nifi 实现这一点？ 我尝试在更新处理器中使用 getStateValue 但这给了我一个静态数字。我的 UpdateAttribute 和 UpdateRecord 如下所示。

Answer 1

使用QueryRecord处理器和处理器支持Apache-Calcite SQL language。

使用ROW_NUMBER()窗口子句将new dynamic property添加到QueryRecord处理器。

示例：

select *,ROW_NUMBER() over(<optional orderby clause>) as seq from FLOWFILE

在查询记录处理器中定义 RecordReader 和 RecordWriter 控制器服务，以在 avro schema 中包含 seq column。

queryrecord 处理器的输出流文件现在将包含 seq column 到流文件中。