How to list S3 objects in parallel in PySpark using flatMap()?

Viewed 1689

I have a dataframe where each row contains a prefix that points to a location in S3. I want to use flatMap() to iterate over each row, list the S3 objects in each prefix and return a new dataframe that contains a row per file that was listed in S3.

I've got this code:

import boto3
s3 = boto3.resource('s3')

def flatmap_list_s3_files(row):
    bucket = s3.Bucket(row.bucket)
    s3_files = []
    for obj in bucket.objects.filter(Prefix=row.prefix):
        s3_files.append(obj.key)

    rows = []
    for f in s3_files:
        row_dict = row.asDict()
        row_dict['s3_obj'] = f
        rows.append(Row(**row_dict))
    return rows

df = <code that loads the dataframe>
df.rdd.flatMap(lambda x: flatmap_list_s3_files(x))).toDF()

The only problem is that the s3 object isn't pickleable I guess? So I'm getting this error and I'm not sure what to try next:

PicklingError: Cannot pickle files that are not opened for reading

I'm a spark noob so I'm hoping there's some other API or some way to parallelize the listing of files in S3 and join that together with the original dataframe. To be clear, I'm not trying to READ any of the data in the S3 files themselves, I'm building a table that is essentially a metadata catalogue of all the files in S3. Any tips would be greatly appreciated.

1 Answers

you can't send an s3 client around your spark cluster; you need to share all the information needed to create one and instantiate it at the far end. I don't know about .py but in the java APIs you'd just pass the path around as a string and then convert that to a Path object, call Path.getFileSystem() and work on there. The Spark workers will cache the Filesystem instances for fast reuse

Related