如何为各个流文件指定优先级属性？

Question

我需要在NiFi中使用PrioritizeAttributePrioritizer.

我已经观察到下面的优先顺序参考. https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#settings

如果我收到10个流文件,那么我需要为每个流文件设置唯一的优先级值.

之后,指定队列配置必须是PrioritizeAttributePrioritizer.

然后根据优先级值处理流文件.

如何设置单独流文件的优先级值或Nifi中的优先级为我的情况？

Answer 1

PriorityAttributePrioritizer通过查找名为"priority"的流文件属性并根据优先级的值按字典顺​​序对流文件进行排序来对流文件进行优先级排序.

您可以使用UpdateAttribute处理器设置优先级属性.例如,如果您有三个逻辑数据源,并且Feed#1是最重要的,则Feed#2是第二重要的,而Feed#3是第三个,那么您可以使用三个UpdateAttribute处理器将priority属性设置为1,2 ,然后使用漏斗将它们全部融合在一起.

您可以在漏斗和下一个处理器之间的队列上设置PriorityAttributePrioritizer,此时,只要优先级= 1的流文件到达队列,它将始终在优先级= 2且优先级= 3的任何流文件之前处理.

确定如何设置优先级实际上取决于您的数据.它通常基于有关数据的某些内容,例如每个流文件中的字段被提取到属性以告诉它优先级,或者只知道来自源#1的所有内容都优先于来自源#2的内容. .设置随机的唯一优先级并不是很有意义,因为你甚至不知道你在优先考虑的是什么.

Answer 2

如果文件是在它们生成的时间之后命名的(例如file_2017-03-03T010101.csv),您是否考虑使用UpdateAttributes将文件名解析为日期,将该日期解析为Epoch(恰好是一个递增的数字)作为第一级索引/优先级？

这样你就可以:

GetFile (single thread) - Connector with FIFO - > UpdateAttribute (adding Epoch from filename date) - Connector with PriorityAttributePrioritizer - > 你的剩余流量

假设文件名是file_2017-03-03T010101.csv,表达式语言将类似于:

${filename:toDate("'file_'yyyy-MM-dd'T'HHmmss'.csv'", "UTC"):toNumber()}