NIFI（S3 到 HDFS）

Question

我正在尝试使用 ListS3、fetchS3object 和 PutHDFS 将数据从 S3 复制到 HDFS。 S3 Bucket中的数据结构如下，需要复制到HDFS，文件夹结构相同（文件夹名是动态的）。

bucketname/parent-folder1/subfolder1/filename1.txt

bucketname/parent-folder1/subfolder2/filename2.txt

bucketname/parent-folder2/subfolder1/filename3.txt

PutHDFS 处理器显示以下错误

org.apache.nifi.processor.exception.ProcessException：将文件复制到 HDFS 但无法重命名点文件 /dev/.parent-folder1/subfolder1/filename1.txt 到它的最终文件名

我了解文件夹在 S3 中是虚拟的。它通过引入 UpdateAttribute 处理器 (${filename:replaceAll("/", "-")}) 来工作，但 structred 文件夹不是在 HDFS 中创建的。还有什么其他选择？有模板吗？

关于错误处理的一些疑惑
1)ListS3 处理器保持状态。当 ListS3 和 fetchS3object 成功而 PutHDFS 失败时会发生什么？将 ListS3 再次加载文件，否则由开发人员处理异常。是否可以重用 fetchS3object 加载的流文件。
2)最终用户如何知道复制成功和复制失败

谢谢 提拉克

Answer 1

我认为问题在于来自 FetchS3Object 的流文件的“filename”属性设置为“parent-folder1/subfolder1/filename1.txt”之类的东西，但 PutHDFS 需要这个值只是“filename1.txt”。 txt”。

您可以通过停止 PutHDFS 并等待流文件进入传入队列来检查这一点，然后列出队列并查看流文件的属性并查看文件名等于什么。

如果我的建议是真的，那么您可以在 PutHDFS 之前使用 UpdateAttribute 来使 filename = ${filename:substringAfterLast('/')}。

然后在 PutHDFS 中创建 Directory 属性“/dev/${path}”或任何具有来自存储桶的路径的属性。

回答您的具体问题...

1) ListS3 不知道它下游的东西是否正常工作，因此它不会重试或重置状态。

2) 你应该根据 PutHDFS 的关系知道成功或失败。您应该在某处路由失败关系，以便可以重新处理或重试。