Hello guys, a question regarding large dataset as artifact. I’m building a demo with metaflow and a few steps are about converting data and splitting it later into test|train sets. Right now I’m thinking to use metaflow.S3 to store output results of a parallel conversion step and then read it in another step. Despite that, looking into docs, there are a few notes that Datastore should be enough for the most cases. I think my case should be under that “most cases”. I wonder how metaflow devs approach this issue. Increasing resources isn’t feasible, since a dataset can be extremely large. So,

join

will be OOM. Is there a way to pass references within join, so it doesn’t load data? Can I then read data iteratively?

How large is the large dataset?

large enough to not consider allocating an instance with sufficient RAM. Let’s say a few dozens GB

If you are using

merge_artifacts

in the

join

step to pass the variables down, it should work without any issues.

it didn’t 🙂 not that long ago we had a discussion here - let’s take CIFAR10, upscale it to 299x299 and try to pass it between steps. Sequential steps do work. Join though, blows up. I tried 24GB of ram if I recall correctly. Anyway, I’m rather looking for a way, how not to load everything into the memory at once. If I can use a generator with

yield

where it pulls data only at the actual yielding

is

result.converted

a big object in the preceding step?

define big 🙂 it’s a list of 1000 upscaled images 299x299

@resources(cpu=2)
    @step
    def convert(self):
        from PIL import Image
        from hxml.constants import DEFAULT_IMAGE_SIZE
        import io
        import numpy as np

        def resize_image(image):
            import tensorflow as tf
            return tf.image.resize(image, DEFAULT_IMAGE_SIZE)

        def convert_to_PIL(image):
            img = resize_image(image)
            # print(f'Generating {img.shape} based on {i}: {d.shape}')
            pimg = Image.fromarray(np.uint8(img))
            imgByteArr = io.BytesIO()
            pimg.save(imgByteArr, format='PNG')
            return imgByteArr.getvalue()

        # print(f'Converting {len(self.input)} input')
        self.converted = [convert_to_PIL(image) for image in self.input]
        self.next(self.join)

If you don't want to use actual data anywhere downstream, you can even not refer to them in the

join

step

or I don’t split them all at once. I can just take chunks , shuffle them and split the chunks. Still, there should be a point where I join them into something. So I can fit the model

It seems that a single image is no more than half a megabyte - so roughly half GB from each foreach split. How many foreach splits do you have?

so, let’s count

@resources(cpu=2)
    @step
    def schedule_conversion(self):
        import numpy as np
        self.chunks = np.split(self.X, len(self.X) // 5000, axis=0)
        self.next(self.convert, foreach='chunks')

60000 total by 5000 = 12

each

convert

step was processing 1000 images?

ah sorry, my bad. Yes, 5000. Got lost in my attempts

Yep

and, how can I overcome it? 🙂

depends on how you want to use

Xnew

further downstream in your flow

a dummy version was to use it for the split

self.X_train, self.X_test, self.y_train, self.y_test = \
            train_test_split(self.Xnew, self.y, test_size=0.2,
                             random_state=42, stratify=self.y)

And this step will run on an instance that has 32GB memory?

it might, but that’s my question - is there a design pattern for such scenario. Allocating 32gb instance for a split, or any other type of processing doesn’t sound right. As I mentioned above, there are multiple ways to split data without pulling it all at once. At the same time, let’s imagine there will be a next step, like training. Where I would like to use batches. At the moment I doubt that I can use artifacts as is

rather than joining together the image bytes into a single array (and loading it all into memory at once to do so), could you pass the pointers to the S3 URLS to defer loading? Depending on your use-case, you could also potentially do the train/test splits relative to the URLs without loading the image bytes

s3 URL was my actual thing today. We’re building an ML platform where Metaflow serves as an ETL. At the end, the first pipeline won’t have any training, just data processing - like rescaling images, converting them from something to something - think like PSD->png. And finishing with an actual dataset of TF Records. Then this will be fed to Sagemaker where we have actual train/tune scripts. We want to have preprocessed data stored for later retrospection, since our data will be regularly updated and we need to perform dedup and other technics to keep models fine. Of course, we want to retrospect train/test split as well. So, it’s more like a production scale than just an experiments pipeline

S3 is certainly a great place for storing image data! It sounds like one of the largest concerns with the batch processing workflow you’ve described would be the variation in file sizes say you’re wanting to batch process 1000 images as you’ve described, and each of those images is a PSD file that varies in size from 1MB-2GB. If the goal is to efficiently utilize compute resources within those constraints, you would need to efficiently be able to construct those batches of 1000 images such that they’re similar in aggregate size across batches. from those 1000 example PSD files, you’d have a lower bound of 1GB (1k 1MB files) up to 2TB (1k 2GB files). The mean/median file size will be somewhere in the middle, and you could look at the distribution of file sizes and try to be clever with batch sampling and/or process them serially within batches to cap the upper bound – but the main thing that seems to stand out is that you’ll want to store the files in S3 (which you could do via metaflow or manually) and then reference pointers to the S3 image urls for tracking the lineage, splits, etc

yes, that’s the plan for now. And files won’t stay in Metaflow s3, they will be transferred at the end of the pipeline to an appropritate datastore. But yes,thank you for confirmation