CODEWITHSUNDEEP
pythongoogle-cloud-storagegoogle-cloud-dataflowapache-beam
edcys
edcys

Reputation: 143

How to read blob (pickle) files from GCS in a Google Cloud DataFlow job?

I try to run a DataFlow pipeline remotely which will use a pickle file. Locally, I can use the code below to invoke the file.

with open (known_args.file_path, 'rb') as fp:
     file = pickle.load(fp)

However, I find it not valid when the path is about cloud storage(gs://...):

IOError: [Errno 2] No such file or directory: 'gs://.../.pkl'

I kind of understand why it is not working but I cannot find the right way to do it.

Upvotes: 2

Views: 6117

Answers (2)

Marcin Zablocki
Marcin Zablocki

Reputation: 10693

If you have pickle files in your GCS bucket, then you can load them as BLOBs and process them further like in your code (using pickle.load()):

class ReadGcsBlobs(beam.DoFn):
    def process(self, element, *args, **kwargs):
        from apache_beam.io.gcp import gcsio
        gcs = gcsio.GcsIO()
        yield (element, gcs.open(element).read())


# usage example:
files = (p
         | "Initialize" >> beam.Create(["gs://your-bucket-name/pickle_file_path.pickle"])
         | "Read blobs" >> beam.ParDo(ReadGcsBlobs())
        )

Upvotes: 4

jkff
jkff

Reputation: 17913

open() is the standard Python library function that does not understand Google Cloud Storage paths. You need to use the Beam FileSystems API instead, which is aware of it and of other filesystems supported by Beam.

Upvotes: 1

Related Questions