我需要将大量文件从同一集群中的一个hdfs目录移动(而不是复制)到另一个hdfs-dir。
我可以使用distcp，但这似乎太过分了，因为它正在复制文件(复制它)。我只想移动它。两个问题：
A)外面有什么东西吗：
我想使用mapreduce来做到这一点，因为有数百万个文件需要移动(或重命名到新路径)。我也想把它和oozie结合起来。我可以自己写一份地图还原工作，但我想知道是否有什么东西可以做这项工作。
B)我真的需要像mapred那样做吗
不幸的是，我对hdfs-rename的性能特征了解不够；你认为我能用单线程方法重命名文件吗？
移动本身是高效的，因为它只在元数据(即inode)级别，而不是在数据级别。换句话说，发出一个移动(在Hadoop的内部代码中称为rename，而不是move)比复制数据快得多。如果您对详细信息感兴趣，可以查看源代码。
因此，不应该执行distcp，因为这将是数据的实际副本。如果你想将其并行化(因为你谈论的是数百万个文件)，那么使用hadoop流：应该不会太难
- 编写多个文件，其中包含要重命名的文件列表(src+destination)，每行一个
- 编写一个shell脚本，为它在stdin上读取的每一行发出重命名(hdfs命令
mv) - 使用流：你的文件和文件是输入，你的shell脚本是映射器
外面有什么吗？
我不知道，但可能有.
我真的需要像映射一样这样做吗？
如果您有数百万个文件，即使HDFS重命名本身很有效，联系namenode的延迟也会增加但是，如果是一次性的，我宁愿发布单线程方法并等待，因为编写和调试(即使是简单的代码)也需要一段时间。如果你计划经常这样做(为什么？)，那么我会考虑实施我上面描述的方法。
如果你想在HDFS:中将文件的子集从一个文件夹复制到另一个文件夹，我会想到这个：
import pandas as pd
import os
from multiprocessing import Process
from subprocess import Popen, PIPE
hdfs_path_1 = '/path/to/the/origin/'
hdfs_path_2 = '/path/to/the/destination/'
df = pd.read_csv("list_of_files.csv")
to_do_list = list(df.tar) # or any other lists that you have
print(f'To go: {len(to_do_list)}')
def copyy(f):
process = Popen(f'hdfs dfs -mv {hdfs_path_1}{f} {hdfs_path_2}', shell=True, stdout=PIPE, stderr=PIPE)
std_out, std_err = process.communicate()
if std_out!= b'':
print(std_out)
ps = []
for f in to_do_list:
p = Process(target=copyy, args=(f,))
p.start()
ps.append(p)
for p in ps:
p.join()
print('done')
此外，如果你想在一个目录中列出所有文件，请使用这个：
from subprocess import Popen, PIPE
hdfs_path = '/path/to/the/designated/folder'
process = Popen(f'hdfs dfs -ls -h {hdfs_path}', shell=True, stdout=PIPE, stderr=PIPE)
std_out, std_err = process.communicate()
list_of_file_names = [fn.split(' ')[-1].split('/')[-1] for fn in std_out.decode().readlines()[1:]][:-1]
list_of_file_names_with_full_address = [fn.split(' ')[-1] for fn in std_out.decode().readlines()[1:]][:-1]