我有一組.hdf文件,我想分發和使用PySpark磚環境下閱讀工作節點上。我能夠閱讀.hdf工人節點上文件和從文件中獲取數據。
現在每個工作節點的下一個要求是必須修改各自的.hdf文件,最後更新的文件應該保存dbfs位置。然而,我無法修改.hdf文件的內容。通過h5py .hdf文件讀取和寫入。.hdf文件的內容保存在工人,但是當我從職工dbfs複製文件,更改不再可見。任何建議將高度讚賞。
這是短的代碼
def change_to_hdf2 (dname、file_name、數據):numpy作為np h5py導入,導入操作係統,與h5py子流程。文件(file_name, a) f: dset = f。create_dataset (dname, data =數據)p =子流程。Popen (“cp {0} {1}”。形式at(os.path.join(os.getcwd(), file_name), os.path.join('/dbfs/mnt', file_name)), stdout=subprocess.PIPE, shell=True) #copy from worker to dbfs return 'Copied back' def create_hdf_file2(file): import numpy as np import h5py, os, subprocess import pandas as pd p = subprocess.Popen("cp {0} {1}".format(file, os.getcwd()), stdout=subprocess.PIPE, shell=True) #copy hdf files from dbfs to worker's directory dbfs_base_path, dbfs_fname = os.path.split(file) #separate file name *****_data_updated = [111,222,333,444,555] df_data2_updated = pd.DataFrame(*****_data_updated, columns=['Numbers']) change_to_hdf2('test_dataset_44', dbfs_fname, df_data2_updated) return True def read_hdf_file2(file_name, dname): import numpy as np import h5py, os, subprocess import pandas as pd with h5py.File(file_name, 'r') as f: data = f[dname] print(data[:5]) return data #main code file_name = ['/dbfs/mnt/file1.hdf', '/dbfs/mnt/file2.hdf'] rdd = spark.sparkContext.parallelize(file_name) result = rdd.map(lambda x: create_hdf_file(x)).collect() print(result) #----------------to read files from the dbfs/mnt after copying files from worker #Test code read_hdf_file('/dbfs/mnt/file1.hdf', dname = 'default') # works as 'default' existed in original file read_hdf_file('/dbfs/mnt/file1.hdf', dname = 'test_dataset_44') # failed, #in above line, 'test_dataset_44' is not being found on dbfs/mnt/file1.hdf
