BUG: Using Modin objects within an `apply` fails, with unclear error message

[zmbc]

Modin version checks

• [X] I have checked that this issue has not already been reported.

• [X] I have confirmed this bug exists on the latest released version of Modin.

• [X] I have confirmed this bug exists on the main branch of Modin. (In order to do this you can follow this guide.)

Reproducible Example

import modin.pandas as pd

abbreviations = pd.Series(['Major League Baseball', 'National Basketball Association'], index=['MLB', 'NBA'])

teams = pd.DataFrame({'name': ['Mariners', 'Lakers'] * 500, 'league_abbreviation': ['MLB', 'NBA'] * 500})

print(teams.set_index('name').league_abbreviation.apply(lambda abbr: abbreviations.loc[abbr]).rename('league'))

Issue Description

This code fails because it tries to use the Modin Series abbreviations within a .apply function.

Running on either Ray or Dask, the error you get is pretty cryptic -- it's not easy to tell that this is the problem.

Expected Behavior

Ideally, it would behave like Pandas, allowing the use of a Modin object within an apply. If this is not possible, at least a nicer error message would help.

Error Logs

With Ray:

```python-traceback Traceback (most recent call last): File "", line 1, in File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/series.py", line 393, in __repr__ temp_df = self._build_repr_df(num_rows, num_cols) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/base.py", line 261, in _build_repr_df return self.iloc[indexer]._query_compiler.to_pandas() File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/storage_formats/pandas/query_compiler.py", line 282, in to_pandas return self._modin_frame.to_pandas() File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/utils.py", line 501, in run_f_on_minimally_updated_metadata result = f(self, *args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/dataframe.py", line 4015, in to_pandas df = self._partition_mgr_cls.to_pandas(self._partitions) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition_manager.py", line 694, in to_pandas retrieved_objects = cls.get_objects_from_partitions(partitions.flatten()) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition_manager.py", line 903, in get_objects_from_partitions return cls._execution_wrapper.materialize( File "~/src/modin/modin/core/execution/ray/common/engine_wrapper.py", line 92, in materialize return ray.get(obj_id) File "~/mambaforge/envs/modin/lib/python3.10/site-packages/ray/_private/auto_init_hook.py", line 24, in auto_init_wrapper return fn(*args, **kwargs) File "~/mambaforge/envs/modin/lib/python3.10/site-packages/ray/_private/client_mode_hook.py", line 103, in wrapper return func(*args, **kwargs) File "~/mambaforge/envs/modin/lib/python3.10/site-packages/ray/_private/worker.py", line 2493, in get raise value.as_instanceof_cause() ray.exceptions.RayTaskError(TypeError): ray::_apply_list_of_funcs() (pid=781021, ip=10.158.106.10) At least one of the input arguments for this task could not be computed: ray.exceptions.RayTaskError: ray::_apply_list_of_funcs() (pid=781021, ip=10.158.106.10) File "~/src/modin/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py", line 421, in _apply_list_of_funcs partition = func(partition, *args, **kwargs) File "~/src/modin/modin/core/dataframe/algebra/map.py", line 51, in lambda x: function(x, *args, **kwargs), *call_args, **call_kwds File "~/mambaforge/envs/modin/lib/python3.10/site-packages/pandas/core/frame.py", line 10123, in map return self.apply(infer).__finalize__(self, "map") File "~/mambaforge/envs/modin/lib/python3.10/site-packages/pandas/core/frame.py", line 10037, in apply return op.apply().__finalize__(self, method="apply") File "~/mambaforge/envs/modin/lib/python3.10/site-packages/pandas/core/apply.py", line 837, in apply return self.apply_standard() File "~/mambaforge/envs/modin/lib/python3.10/site-packages/pandas/core/apply.py", line 963, in apply_standard results, res_index = self.apply_series_generator() File "~/mambaforge/envs/modin/lib/python3.10/site-packages/pandas/core/apply.py", line 979, in apply_series_generator results[i] = self.func(v, *self.args, **self.kwargs) File "~/mambaforge/envs/modin/lib/python3.10/site-packages/pandas/core/frame.py", line 10121, in infer return x._map_values(func, na_action=na_action) File "~/mambaforge/envs/modin/lib/python3.10/site-packages/pandas/core/base.py", line 921, in _map_values return algorithms.map_array(arr, mapper, na_action=na_action, convert=convert) File "~/mambaforge/envs/modin/lib/python3.10/site-packages/pandas/core/algorithms.py", line 1814, in map_array return lib.map_infer(values, mapper, convert=convert) File "lib.pyx", line 2917, in pandas._libs.lib.map_infer File "~/src/modin/modin/pandas/series.py", line 1222, in lambda s: arg(s) File "", line 1, in File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/indexing.py", line 656, in __getitem__ return self._helper_for__getitem__( File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/indexing.py", line 703, in _helper_for__getitem__ result = self._get_pandas_object_from_qc_view( File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/indexing.py", line 399, in _get_pandas_object_from_qc_view return res_df.squeeze(axis=axis) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/series.py", line 1808, in squeeze return self._reduce_dimension(self._query_compiler) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/series.py", line 2265, in _reduce_dimension return query_compiler.to_pandas().squeeze() File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/storage_formats/pandas/query_compiler.py", line 282, in to_pandas return self._modin_frame.to_pandas() File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/utils.py", line 501, in run_f_on_minimally_updated_metadata result = f(self, *args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/dataframe.py", line 4015, in to_pandas df = self._partition_mgr_cls.to_pandas(self._partitions) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition_manager.py", line 694, in to_pandas retrieved_objects = cls.get_objects_from_partitions(partitions.flatten()) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition_manager.py", line 903, in get_objects_from_partitions return cls._execution_wrapper.materialize( File "~/src/modin/modin/core/execution/ray/common/engine_wrapper.py", line 92, in materialize return ray.get(obj_id) ray.exceptions.RayTaskError(TypeError): ray::_apply_func() (pid=781713, ip=10.158.106.10) File "~/src/modin/modin/core/execution/ray/implementations/pandas_on_ray/partitioning/partition.py", line 379, in _apply_func result = func(partition, *args, **kwargs) TypeError: 'NoneType' object is not callable ```

With dask:

```python-traceback Traceback (most recent call last): File "", line 1, in File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/series.py", line 393, in __repr__ temp_df = self._build_repr_df(num_rows, num_cols) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/base.py", line 261, in _build_repr_df return self.iloc[indexer]._query_compiler.to_pandas() File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/storage_formats/pandas/query_compiler.py", line 282, in to_pandas return self._modin_frame.to_pandas() File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/utils.py", line 501, in run_f_on_minimally_updated_metadata result = f(self, *args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/dataframe.py", line 4015, in to_pandas df = self._partition_mgr_cls.to_pandas(self._partitions) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition_manager.py", line 694, in to_pandas retrieved_objects = cls.get_objects_from_partitions(partitions.flatten()) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition_manager.py", line 903, in get_objects_from_partitions return cls._execution_wrapper.materialize( File "~/src/modin/modin/core/execution/dask/common/engine_wrapper.py", line 109, in materialize return client.gather(future) File "~/src/modin/modin/core/execution/dask/implementations/pandas_on_dask/partitioning/partition.py", line 358, in apply_list_of_funcs partition = func(partition, *f_args, **f_kwargs) File "~/src/modin/modin/core/dataframe/algebra/map.py", line 51, in lambda x: function(x, *args, **kwargs), *call_args, **call_kwds File "lib.pyx", line 2917, in pandas._libs.lib.map_infer File "~/src/modin/modin/pandas/series.py", line 1222, in lambda s: arg(s) File "", line 1, in File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/indexing.py", line 656, in __getitem__ return self._helper_for__getitem__( File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/pandas/indexing.py", line 702, in _helper_for__getitem__ qc_view = self.qc.take_2d_labels(row_loc, col_loc) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/storage_formats/base/query_compiler.py", line 4135, in take_2d_labels return self.take_2d_positional(row_lookup, col_lookup) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/storage_formats/pandas/query_compiler.py", line 4277, in take_2d_positional self._modin_frame.take_2d_labels_or_positional( File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/utils.py", line 501, in run_f_on_minimally_updated_metadata result = f(self, *args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/dataframe.py", line 954, in take_2d_labels_or_positional return self._take_2d_positional(row_positions, col_positions) File "~/src/modin/modin/logging/logger_decorator.py", line 129, in run_and_log return obj(*args, **kwargs) File "~/src/modin/modin/core/dataframe/pandas/dataframe/dataframe.py", line 1149, in _take_2d_positional [ File "~/src/modin/modin/core/dataframe/pandas/dataframe/dataframe.py", line 1150, in [ File "~/src/modin/modin/core/dataframe/pandas/dataframe/dataframe.py", line 1151, in self._partitions[row_idx][col_idx].mask( File "~/src/modin/modin/core/execution/dask/implementations/pandas_on_dask/partitioning/partition.py", line 177, in mask new_obj = super().mask(row_labels, col_labels) File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition.py", line 257, in mask new_obj = self.add_to_apply_calls(self._iloc_func, row_labels, col_labels) File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition.py", line 162, in add_to_apply_calls return self.__constructor__( File "~/src/modin/modin/core/execution/dask/implementations/pandas_on_dask/partitioning/partition.py", line 46, in __init__ super().__init__() File "~/src/modin/modin/core/dataframe/pandas/partitioning/partition.py", line 55, in __init__ self.execution_wrapper.put(self._iloc) File "~/src/modin/modin/core/execution/dask/common/engine_wrapper.py", line 135, in put client = default_client() File "~/mambaforge/envs/modin/lib/python3.10/site-packages/distributed/client.py", line 5550, in default_client raise ValueError( ValueError: No clients found Start a client and point it to the scheduler address from distributed import Client client = Client('ip-addr-of-scheduler:8786') ```

Installed Versions

INSTALLED VERSIONS ------------------ commit : 513166faea5926101ad14ecc149b6c72d3376866 python : 3.10.12.final.0 python-bits : 64 OS : Linux OS-release : 5.4.0-135-generic Version : #152-Ubuntu SMP Wed Nov 23 20:19:22 UTC 2022 machine : x86_64 processor : x86_64 byteorder : little LC_ALL : None LANG : en_US.UTF-8 LOCALE : en_US.UTF-8 Modin dependencies ------------------ modin : 0.23.0+107.g513166fa ray : 2.6.1 dask : 2023.7.1 distributed : 2023.7.1 hdk : None pandas dependencies ------------------- pandas : 2.1.1 numpy : 1.25.1 pytz : 2023.3 dateutil : 2.8.2 setuptools : 68.0.0 pip : 23.2.1 Cython : None pytest : 7.4.0 hypothesis : None sphinx : 7.1.0 blosc : None feather : 0.4.1 xlsxwriter : None lxml.etree : 4.9.3 html5lib : None pymysql : None psycopg2 : 2.9.6 jinja2 : 3.1.2 IPython : 8.14.0 pandas_datareader : None bs4 : 4.12.2 bottleneck : None dataframe-api-compat: None fastparquet : 2022.12.0 fsspec : 2023.6.0 gcsfs : None matplotlib : 3.7.2 numba : None numexpr : 2.8.4 odfpy : None openpyxl : 3.1.2 pandas_gbq : 0.15.0 pyarrow : 12.0.1 pyreadstat : None pyxlsb : None s3fs : 2023.6.0 scipy : 1.11.1 sqlalchemy : 1.4.45 tables : 3.8.0 tabulate : None xarray : None xlrd : 2.0.1 zstandard : None tzdata : 2023.3 qtpy : 2.3.1 pyqt5 : None

[mvashishtha]

@zmbc thank you for reporting this issue. I can reproduce it at version ea8088af4cadfb76294e458e5095f262ca85fea9. Indeed, you should assume that your callables will apply to pandas objects as opposed to modin.pandas objects, and you should avoid referencing modin.pandas objects within the callables. This is an important bug that we should document and possibly even warn users about the first time they use a method like apply or map.

Even if we did allow access to modin.pandas objects inside the callables, the performance would probably not be good because we'd have to pass the entire dataframe to every partition that's accessing the dataframe. For a distributed dataframe, you want to avoid ever materializing all the data in one worker. Better would be to rewrite the code so it's more amenable to distribution, e.g. for your example:

teams.set_index('name').merge(abbreviations.to_frame(), left_on='league_abbreviation', right_index=True, how='left')[0].rename('league')

alternatively, if the list of abbrevations is really small, use df.replace with a dictionary of abbreviations:

teams.set_index('name').league_abbreviation.replace({'MLB': 'Major League Baseball', "NBA": 'National Basketball Association'})

Despite what I've written above, the error you get here isn't helpful at all. I'll investigate to see why Modin raises this error and whether we can either fix the bug or give a more informative error.

(by the way, a note that @zmbc might not find useful but that might be useful for me and future modin maintainers: this works fine in MODIN_ENGINE=PYTHON. Seems like it's a ray/dask problem with serializing modin frames for partitions)

[anmyachev]

@mvashishtha Maybe changing __reduce__ function will help us here?

Something like:

    @classmethod
    def _inflate_full(cls, pandas_series, source_pid):
        """
        Re-creates the object from previously-serialized disk-storable representation.

        Parameters
        ----------
        pandas_series : pandas.Series
            Data to use for object re-creation.

        Returns
        -------
        Series
            New Series based on the `pandas_series`.
        """
        import os
        current_pid = os.getpid()
        print(f"Current pid: {current_pid}")
        if current_pid != source_pid:
            # Pandas objects can be used in any processes.
            return pandas_series
        # Modin objects can only be created in the main process.
        return cls(data=pandas_series)

    def __reduce__(self):
        import os
        self._query_compiler.finalize()
        pid = os.getpid()
        print(f"Source pid: {pid}")
        # there is can be the flag in Series itself:
        # for example, self._make_full_reduce
        if True:  # PersistentPickle.get():
            return self._inflate_full, (self._to_pandas(), pid)
        return self._inflate_light, (self._query_compiler, self.name)

[zmbc]

Even if we did allow access to modin.pandas objects inside the callables, the performance would probably not be good because we'd have to pass the entire dataframe to every partition that's accessing the dataframe. For a distributed dataframe, you want to avoid ever materializing all the data in one worker.

I'm not so sure that this is a fundamental limitation. If the callable could directly use modin.pandas objects, it could do some operation on them, such as a .loc, that greatly reduced the data size before it needed to be materialized. Sure, it could do some operation using an entire Modin dataframe, and that would be slow, but I don't think anyone would be surprised that it was slow.

My example is obviously contrived and could have been done much better with join or merge. But I do think this functionality would be useful to have in general.

[anmyachev]

Even if we did allow access to modin.pandas objects inside the callables, the performance would probably not be good because we'd have to pass the entire dataframe to every partition that's accessing the dataframe. For a distributed dataframe, you want to avoid ever materializing all the data in one worker.

I'm not so sure that this is a fundamental limitation. If the callable could directly use modin.pandas objects, it could do some operation on them, such as a .loc, that greatly reduced the data size before it needed to be materialized. Sure, it could do some operation using an entire Modin dataframe, and that would be slow, but I don't think anyone would be surprised that it was slow.

My example is obviously contrived and could have been done much better with join or merge. But I do think this functionality would be useful to have in general.

Agree. We can give performance warnings, but the code should work anyway.