nifi Json data using routeonattributeto 过滤属性或属性值

Question

我目前正在使用 Nifi 使用数据来读取 tealium 事件流并加载到 HDFS 中。当源未发送属性数据时需要帮助过滤数据。

{"account":"newtv","twitter:description":"Discover when your favorite NewTV shows and hosts are being shown. ","og:locale":"en_US","dcterms:publisher":"NewTV","original-source":"www.newtv.com/","og:url":"www.newtv.com/show/program-guide"}},"post_time":"2019-10-09 11:27:46","useragent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36","event_id":"12345"}

以上消息示例。当源未发送来自以下示例数据集的 event_id 属性的数据时，我目前坚持过滤数据。

当前Nifi流量， Consume Kafka -> Evaluate Json Path -> Jolttransform Json -> Evaluate Json Path-> RouteOnAttribute -> Merge Content -> Evaluate Json Path -> 更新属性 -> PutHDFS ->MoveHDFS

需要帮助如何使用 RouteOnAttribute 拆分数据以区分缺少的 event_id 属性或 attribute_value 到两个不同的流。带有属性或属性值和缺失值的流向错误并加载到不同的输出路径。

Answer 1

在 EvaluateJsonPath 处理器中添加新的属性以从流文件中提取 event_id 值。

如果 flowfile 没有 event_id 然后 nifi 将空值添加到属性。

EvaluateJsonPath Configs:

然后通过使用 RouteOnAttribute 处理器，我们可以检查属性值并相应地路由流文件。

RouteOnAttribute Configs:

非空值

${event_id:isEmpty():not()}

空值

${event_id:isEmpty()}

然后使用空值和非空值关系进行进一步处理..!!

nifi Json data using routeonattributeto 过滤属性或属性值

nifi Json data using routeonattributeto filter attribute or attribute value

apache-nifi

cloudera

hortonworks-data-platform

data-ingestion

hortonworks-sam