NiFi：在流文件中的所有记录前附加（或附加）行号

Question

我使用的是 NiFi 1.9.2

我正在阅读一个文本文件，它恰好是一个 csv 文件。我在 flowFile 的内容中有文件的内容。

内容是

a,b,c
d,e,f
g,h,i

我想在流文件中的所有记录前面加上一个行号并获取

1,a,b,c
2,d,e,f
3,g,h,i

每次我通过这个处理器输入文件时

我可以通过使用带有以下属性的 ReplaceText 处理器来实现一些接近的目标：

Search Value : (?m)(^.*$)
Replacement Value : ${nextInt()},$1

但是因为 nextInt() 在我得到的运行 NiFi 实例的整个生命周期内都保持它的价值

    0,a,b,c
    1,d,e,f
    2,g,h,i
for 1st execution

    3,a,b,c
    4,d,e,f
    5,g,h,i
for the next execution etc

此外，根据 NiFi 表达式语言指南，“计数器在所有 NiFi 组件之间共享，因此从一个处理器多次调用此函数将不能保证特定处理器上下文中的顺序值。”

1. 有没有办法确保在 NiFi 实例的生命周期内每次执行此处理器的行号始终从 0 开始，并且始终是连续的？

2. 计数器的范围是多少？

3. 我可以让计数器从 1 开始吗？

Answer 1

您可以将内容拆分为多行，然后使用 fragment.index 将计数器添加到行中。之后，您可以再次合并它们。

流程：

生成流文件：

拆分文本：

替换文本：

合并内容： 不要忘记在 Demarcator 属性中添加一个新行（Shift+Enter）。

结果：

如果你想从零开始计数，你可以使用${Fragment.index:minus(1)}。