如果您想在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]

移动本身是有效的，因为它只在元数据（即inode）级别，而不是在数据级别。换句话说，发出一个move（在hadoop的代码内部称为 rename ，不是 move )比复制数据快得多。你可以看看源代码，以防你对细节感兴趣。
因此，您不应该执行distcp，因为这将是数据的实际副本。如果你想并行化它（因为你说的是数百万个文件），使用hadoop流媒体应该不会太难：
写几个包含要重命名的文件列表的文件（src+destination），每行一个。
编写shell脚本以发出rename（hdfs）命令 mv )它在标准数据上每读一行。
使用流：你的文件和文件是输入，你的shell脚本是Map器。
外面有什么东西吗？
我不知道，但可能有。
我真的需要像Map上那样做吗？
如果您有数百万个文件，那么联系namenode的延迟就会增加，即使hdfs重命名本身是有效的。但是，如果它是一次性的，我宁愿使用单线程方法并等待，因为编写和调试（即使是简单的代码）也需要一段时间。如果您计划经常这样做（为什么？），那么我会考虑实施上面描述的方法。