https://www.getdaft.io logo
Join Slack
Powered by
Hi, I am trying to switch to `daft.cls`, but seein...
# general
g
Hi, I am trying to switch to 
daft.cls
, but seeing the following error (🧵), any ideas?
Copy code
8e5a750a1b5e37ed1885cd6ba358f551e/virtualenv/lib/python3.12/site-packages/daft/runners/ray_runner.py", line 1391, in run_iter
    yield from self.flotilla_plan_runner.stream_plan(
  File "/tmp/ray/session_2025-12-17_22-08-09_629560_1/runtime_resources/pip/c7b0a478e5a750a1b5e37ed1885cd6ba358f551e/virtualenv/lib/python3.12/site-packages/daft/runners/flotilla.py", line 326, in stream_plan
    materialized_result = ray.get(self.runner.get_next_partition.remote(plan_id))
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/_private/auto_init_hook.py", line 21, in auto_init_wrapper
    return fn(args, kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/_private/client_mode_hook.py", line 103, in wrapper
    return func(args, kwargs)
2025-12-19 09:58:41,599 - INFO - ray_job_helper.py:130 -            ^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/_private/worker.py", line 2661, in get
    values, debugger_breakpoint = worker.get_objects(object_refs, timeout=timeout)
                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/_private/worker.py", line 871, in get_objects
    raise value.as_instanceof_cause()
ray.exceptions.RayTaskError(RuntimeError): ray::RemoteFlotillaRunner.get_next_partition() (pid=38966, ip=10.216.131.13, actor_id=052600750315e1d4f096b3601c000000, repr=<daft.runners.flotilla.RemoteFlotillaRunner object at 0x7f14b9c75e20>)
  File "/home/ray/anaconda3/lib/python3.12/concurrent/futures/_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
2025-12-19 09:58:41,599 - INFO - ray_job_helper.py:130 -   File "/home/ray/anaconda3/lib/python3.12/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/tmp/ray/session_2025-12-17_22-08-09_629560_1/runtime_resources/pip/c7b0a478e5a750a1b5e37ed1885cd6ba358f551e/virtualenv/lib/python3.12/site-packages/daft/runners/flotilla.py", line 267, in get_next_partition
    next_partition_ref = await self.curr_result_gens[plan_id].__anext__()
                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/tmp/ray/session_2025-12-17_22-08-09_629560_1/runtime_resources/pip/c7b0a478e5a750a1b5e37ed1885cd6ba358f551e/virtualenv/lib/python3.12/site-packages/daft/execution/ray_actor_pool_udf.py", line 94, in start_udf_actors
    ).remote(expr_projection)
      ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/actor.py", line 833, in remote
2025-12-19 09:58:41,599 - INFO - ray_job_helper.py:130 -     return actor_cls._remote(args=args, kwargs=kwargs, updated_options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
           ^^^^^^^^^^^^^^^^^^^
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/actor.py", line 1161, in _remote
    actor_id = worker.core_worker.create_actor(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^
             ^^^^^^^^^^^^^
2025-12-19 09:58:42,599 - INFO - ray_job_helper.py:130 -   File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/cloudpickle/cloudpickle.py", line 1479, in dumps
    cp.dump(obj)
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/cloudpickle/cloudpickle.py", line 1245, in dump
    return super().dump(obj)
           ^^^^^^^^^^^^^^^^^
RuntimeError: Failed to serialize: OtherString("PicklingError: Cannot pickle files that are not opened for reading: a")
2025-12-19 09:58:47,676 - INFO - ray_job_helper.py:110 - Status for job 'raysubmit_9L8PFC699vLKShv9': FAILED
2025-12-19 09:58:47,677 - INFO - ray_job_helper.py:112 - Job status message: Job entrypoint command failed with exit code 1, last available logs (truncated to 20,000 chars):
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^
             ^^^^^^^^^^^^^
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/cloudpickle/cloudpickle.py", line 1479, in dumps
    cp.dump(obj)
  File "/home/ray/anaconda3/lib/python3.12/site-packages/ray/cloudpickle/cloudpickle.py", line 1245, in dump
    return super().dump(obj)
           ^^^^^^^^^^^^^^^^^
RuntimeError: Failed to serialize: OtherString("PicklingError: Cannot pickle files that are not opened for reading: a")
Copy code
def __init__(self) -> None:
        lock = filelock.FileLock("/tmp/my.lock")
        with lock:
            if not os.path.exists(self.LOCAL_TEXT_CLASSIFICATION_MODEL_PATH):
                self._download_model()
            self.model = self._initialize_model()
c
are you able to provide any more of the udf code? The new 
daft.cls
does a check to make sure that all state is pickleable. The legacy 
daft.udf
did not do this check and it caused some unexpected issues when running distributed. So likely something within your class state is not pickleable.
g
I see, passing in the model class directly as opposed to trying to init within the class I get the error you mention.
Copy code
ValueError: Daft classes must be serializable. If your class accesses a non-serializable global or nonlocal variable, initialize it in the setup method instead.
essentially changing to:
Copy code
def __init__(self, model: CuisineClassificationModel) -> None:
        self.model = model
so 
CuisineClassificationModel
must not be pickable?
that class loads a few things:
Copy code
def __init__(
        self,
        cuisine_model_config: cuisine_classification_model_config.CuisineClassificationModelConfig,
    ) -> None:
        self.tokenizer, self.bert_model, self.device = self._initialize_vectorizer(
            {"vectorizer_config": cuisine_model_config.vectorizer_config}
        )
        self.cuisine_classification_model = tf.saved_model.load(
            cuisine_model_config.text_classification_model["model"]
        )
        with ds_io.open(
            os.path.join(
                cuisine_model_config.text_classification_model["model"],
                "assets/class_labels",
            ),
            "r",
        ) as class_labels:
            self.cuisine_labels = [l.strip() for l in class_labels.readlines()]
5 Views
Open in Slack
PreviousNext
Powered by