Apache Nifi 将文件移动到新的 hdfs 文件夹以获取小于当前日期的文件

Posted 2023-02-16

【中文标题】Apache Nifi 将文件移动到新的 hdfs 文件夹以获取小于当前日期的文件

【英文标题】：Apache Nifi to move files to new hdfs folder for the files lesser than current date

【发布时间】：2019-10-11 22:09:32

【问题描述】：

我正在创建端到端流，通过使用 Kafka 来处理通过 tealium 事件流接收的 Json 文件，从而将数据消耗到 HDFS 中。

目前我用过

Consume Kafka -> Evaluate Json Path -> Jolttransform Json -> Merge Content -> Evaluate Json Path -> Update attribute -> PutHDFS ->MoveHDFS

要求是将全天假脱机的 JSON 数据读取到单个文件中，引用属性 postdate（隐蔽纪元到 YYYYMMDDSS 之前的时间戳）并每天读取数据以合并到单个输出文件中，最后按照以下方式重命名文件与 POST_DATE 字段相关的时间戳以区分每日文件。

当前日期输出文件夹应仅包含当前日期处理文件，所有已完成的较早日期输出文件应移至不同文件夹。

您能否帮助我如何在 MoveHDFS 上递归搜索 hdfs 文件夹并将不等于当前日期的已完成输出文件移动到不同的文件夹中。

【参考方案1】：

当前流程运行良好。 使用 Kafka -> 评估 Json 路径 -> Jolttransform Json -> 合并内容 -> 评估 Json 路径 -> 更新属性 -> PutHDFS ---> 创建合并文件。

上述流程完成后创建一个单独的流程以接收处理后的合并文件并使用 listhdfs->fethchdfs->updateattribute->puthdfs 重新处理它

在 listhdfs 中，设置消费前的最小文件年龄等待时间。这将允许进程递归搜索文件，并使用 updateattribute 根据父文件夹重新创建文件夹以重用进程文件。