如何将numpy数组从PySpark工作程序保存到HDFS或共享文件系统

Question

我想在PySpark中高效地将numpy数组从工作机器（函数）保存/读取到HDFS。我有两台机器A和B。A有师傅和工人。B有一个工人。例如，我想实现以下目标：

if __name__ == "__main__":
    conf = SparkConf().setMaster("local").setAppName("Test")
    sc = SparkContext(conf = conf)
    sc.parallelize([0,1,2,3], 2).foreachPartition(func)
def func(iterator):
    P = << LOAD from HDFS or Shared Memory as numpy array>>
    for x in iterator:
        P = P + x
    << SAVE P (numpy array) to HDFS/ shared file system >>

什么是快速有效的方法？

Answer 1

我偶然发现了同样的问题。最后在Python3.4中使用了HdfsCli模块和tempfiles。

1. 进口：

from hdfs import InsecureClient
from tempfile import TemporaryFile

2. 创建一个hdfs客户端。在大多数情况下，最好在脚本中的某个位置设置一个实用程序函数，如下所示：

def get_hdfs_client():
    return InsecureClient("<your webhdfs uri>", user="<hdfs user>",
         root="<hdfs base path>")

3. 在worker函数中加载并保存您的numpy：

hdfs_client = get_hdfs_client()
# load from file.npy
path = "/whatever/hdfs/file.npy"
tf = TemporaryFile()
with hdfs_client.read(path) as reader:
    tf.write(reader.read())
    tf.seek(0) # important, set cursor to beginning of file
np_array = numpy.load(tf)
...
# save to file.npy
tf = TemporaryFile()
numpy.save(tf, np_array)
tf.seek(0) # important ! set the cursor to the beginning of the file
# with overwrite=False, an exception is thrown if the file already exists
hdfs_client.write("/whatever/output/file.npy", tf.read(),  overwrite=True) 

注：

• 用于创建hdfs客户端的uri以http://开头，因为它使用了hdfs文件系统的web接口
• 确保传递给hdfs客户端的用户具有读写权限
• 根据我的经验，开销并不显著（至少在执行时间方面）
• 使用tempfiles（与/tmp中的常规文件相比）的优点是，无论脚本正常与否，都可以确保集群计算机中没有垃圾文件