Databricks pyspark parallelize unzipping multiple files

user_2023759 0
Viewed 248

I'm trying to parallelize unzipping files stored in s3 in pyspark on Databricks. Unzipping in a for loop works as so:

file_list = [(file.path, file.name) for file in dbutils.fs.ls(data_path) if os.path.basename(file.path).endswith(".zip")] # data_path is taken as a parameter
file_names = [ff[1] for ff in file_list]
for ff in file_list:
  dbutils.fs.cp(ff[0], "/FileStore/tmp/" + ff[1])
  cmd = 'unzip /dbfs/FileStore/tmp/' + ff[1]
  os.system(cmd)
  dbutils.fs.cp("file:/databricks/driver/" + ff[1], data_path)
  dbutils.fs.rm("file:/databricks/driver/" + ff[1])

I'm trying to parallelize the unzip part. So after copying the files to "/FileStore/tmp/" I'm running:

unzips = [file[1] for file in file_list]
def f(x):
  os.system('unzip /dbfs/FileStore/tmp/' + x)
sc.parallelize(unzips).foreach(f)

The job runs but the files are not unzipped anywhere.

pyspark databricks unzip
1 Answers

Just add the -d option to put output onto DBFS, something like this:

def f(x):
  os.system('unzip -o -d /dbfs/FileStore/tmp-output/ /dbfs/FileStore/tmp/' + x)

if necessary (for example, there is no directories inside archive), add more unique directory to each run, for example, append x to it.

Alex_Ott_186270
Related