https://www.getdaft.io logo
Join Slack
Powered by
Hi Guys I'm still struggling a lot to understand ...
# daft-dev
a
Hi Guys I'm still struggling a lot to understand the daft.cls API for model inference if I want to avoid passing hard-coded values. Below I have a simple dummy resnet model that expects as input a batch of images and returns 2 outputs a json string with the probs and the embeddings. More importantly I want to control max_concurrency and batch size during runtime. That means I need to create the class without decorators and then wrap it during runtime. Questions: 1. I know I need to use daft.cls and daft.method.batch but how to do this during runtime when not able to use decorators which expect import time variables; or how to use the decorators but overwrite some of these parameters like max_concurrency and batch_size at runtime. 2. How to mix unested=True when using daft.method.batch which expects output to be a series or a list but unnest-True expects output to be a Struct?
Copy code
class SimpleModel:
    def __init__(self, model_checkpoint_uri: str):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = load_model(model_uri).to(self.device)
        self.model.eval()

    def predict(self, images):
        if len(images) == 0:
            return []

        images_tensor = torch.from_numpy(np.array(images.to_pylist())).to(self.device)
        batch_size = images_tensor.shape[0]
        with torch.inference_mode():
            probabilities, features = self.model(images_tensor)
            features = features.half().detach().cpu().numpy().tolist()
            rounded_probabilities = probabilities.round(decimals=4).detach().cpu().numpy().tolist()
            predicted_class_id = probabilities.argmax(dim=1).detach().cpu().tolist()
            predicted_class_name = [CLASSES[p] for p in predicted_class_id]

            predictions = []
            for i in range(batch_size):
                prediction = json.dumps({
                    "class_id": predicted_class_id[i],
                    "class_name": predicted_class_name[i],
                    "probability": rounded_probabilities[i],
                })
                predictions.append(prediction)

            return predictions, features
During Runtime: Not working
Copy code
ModelClass = daft.cls(EyesClosedClassifier, gpus=1, max_concurrency=cfg.num_gpus)
    model = ModelClass(model_checkpoint_uri=cfg.pipeline.model.checkpoint_uri)
    model_inference = daft.method.batch(
        model.predict,
        return_dtype=daft.DataType.struct({
            "predictions": daft.DataType.string(),
            "features": daft.DataType.list(daft.DataType.float32()),
        }),
        unnest=True,
        batch_size=cfg.batch_size,
    )
r
This will be fun with some higher-order functions, let me make sure I understand correctly. Here's an example: https://gist.github.com/rchowell/7437b2a4e8e8c8c38fa8b9b97c435337 You can do the same thing for classes since these are effectively functions too, your higher-order function just creates the decorated class with captured values instead of a decorated function.
a
@Robert Howell sorry, I'm not following. Is this related to the runtime variables or the multiple outputs issue? Is what you are saying is to put the embeddings in the JSON string and return only 1 output and the create a from_json function that then creates the second column?
r
Let me get you a concrete gist, the JSON thing was just a high-level example of the pattern.
You pass the decorator arguments into a higher-level function which defines the class-based UDF and returns it. The trick is the 
_Extractor().extract
at the end which gives you an Expression type backed by the underlying UDF, which is what allows you to use it in the DSL.
It's the same pattern as the JSON example, just applied to the class-based UDF. Is this helpful, or am I missing something?
a
got it. yeah I want to make it very generic so I can just use the same function/step for many different models, so it can get a bit ugly with a wrapper, but that makes sense. I would be great if we could support the legacy way of having some of these variables modifyiable during runtime so I can still use decorators but modify some properties of it. udf_class.with_batch_size() or udf_class.with_max_concurrency... Either way thanks for the help 🙏!
👍 1
@Robert Howell I managed to get it working thanks. Follow up question if possible. I am running this lightweight model on like 60M+ images. The pipeline is very heavy on pre-processing. I have to get the read image from s3 and read bbox text_file from s3 then crop the object of interest and resize the object and then do inference. I did the test on like 80k images and 16 GPUs and it works great I get decent speed and takes like 2-3 minutes for this number of images. Given the heavy pre-processing it is not a bad number to start without tuning. Currently I use daft v0.7.4 dynamic batch and I do into_partition(16) and max_concurrency=16 which is the num of gpus. I do not see any memory issue when looking at only 80k images but as soon as I remove the df.limit(80000) to run on all 60M+ images I hit CPU OOM very quickly like 3 minutes into the job and actors start to die. Is there something I am doing wrong? Do I need to set the concurrency of cpu steps also? I make sure to delete the image and image_bytes before writing to parquet, so not sure why just increasing the number of row creates OOM issues. the input dataframe is videos which I explode into images since I can parse and have the s3 image url for every image to avoid reading videos...
I'm unable to use daft.resize since I need to crop first and also create a based on the text file.
c
You mention you are exploding into images. Any estimation on the expansion factor (1-N)? like roughly how many (N) rows are being produced?
a
I have a about 300 images per rows so the explosion is from like 200k video rows to 60M images. But I explode before doing any heavy compute or even reading anything from s3.
I do have a ugly function after downloading and decode which takes a daft image convert it to pil to do the cropping and resizing and then return it as a uint8 numpy before inference
c
I am assuming that its because the udfs are now getting much larger batches than before the limit, so that's causing the OOM. in the udf, you can manually set the batch size. You could try adjusting that. We also have a new dynamic batching config that'll dynamically adjust the batch sizing for you.
Copy code
with daft.execution_config_ctx(enable_dynamic_batching=True):
  // df = ...
a
Yes I'm using the dynamic batch size enabled. This preprocessing steps use daft.func . I can try changing it to a method.batch and manually control the batch size of the cpu ops.
I tried daft.execution_config_ctx(enable_dynamic_batching=True) but still getting OOM. I make sure to do to 
into_partitions(16)
after the explode. I also tried removing the expensive preprocessing and just use daft.resize just testing purposed but still gettin OOM. I also try to manually set 
into_batches(10)
before image download and decode but as soon as I do that I start getting too many spills and eventually get object lost error. Running out of ideas 😕
c
Could you try unsetting the dynamic_batching flag, and setting the batch size so something really small like 
1
or 
2
just to see if it is in fact the udf OOMing.
Copy code
model_inference = daft.method.batch(
        model.predict,
        return_dtype=daft.DataType.struct({
            "predictions": daft.DataType.string(),
            "features": daft.DataType.list(daft.DataType.float32()),
        }),
        unnest=True,
        batch_size=1
    )
a
@Cory Grinstead ok. I'll do that a report back. So far I decided to dump the table into a parquet after explosion to isolate explosion being the issue, but that did not help. I'm currently trying to read the parquet after explosion, so it's a 60M rows parquet. I'm mostly doing: 1. df =df.read_parquet (60M rows with urls and meta-data) 2. df = df.into_partitions(16) 3. a few lightweight df.funcs and filter out the nulls of these lighweight funcs. 4. df = df.into_batches(256) 5. heavy daft.funcs (read image, resize, read txt...) aplly null filters 6. model inference with batch_size=128 This is after disabling the dynamic flag.
e
You could use daft.File to lazily load images into gpu memory. read, resize, infer, all in one udf.
a
Yes that's another option. I would also have to load the boxes from s3 inside.
🙌 1
2 Views
Open in Slack
PreviousNext
Powered by