Hi. I am going through the examples of preprocessor from official docs. Everything worked fine if I ran the code in a local Ray instance. However, things went wrong when I tried to execute the code on a remote cluster using Ray Client.
import pandas as pd
import ray
from ray.data.preprocessors import Normalizer
ray.init(address="ray://192.168.154.137:10001", runtime_env={"working_dir": "./"})
df = pd.DataFrame({"X1": [1, 1], "X2": [1, 0], "X3": [0, 1]})
ds = ray.data.from_pandas(df)
ds.to_pandas()
preprocessor = Normalizer(columns=["X1", "X2"])
preprocessor.fit_transform(ds).to_pandas()
The code above is copied from the official doc. I just added one line to connect to a remote cluster. Then I get errors below:
2023-12-13 09:02:54,562 INFO packaging.py:530 -- Creating a file package for local directory './'.
2023-12-13 09:02:54,570 INFO packaging.py:358 -- Pushing file package 'gcs://_ray_pkg_2ecba2b640a55ee8.zip' (0.06MiB) to Ray cluster...
2023-12-13 09:02:54,574 INFO packaging.py:371 -- Successfully pushed file package 'gcs://_ray_pkg_2ecba2b640a55ee8.zip'.
SIGTERM handler is not set because current thread is not the main thread.
2023-12-13 09:02:58,244 INFO streaming_executor.py:104 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(Normalizer._transform_pandas)]
2023-12-13 09:02:58,244 INFO streaming_executor.py:105 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=False, actor_locality_enabled=True, verbose_progress=False)
2023-12-13 09:02:58,244 INFO streaming_executor.py:107 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running: 0.0/18.0 CPU, 0.0/0.0 GPU, 0.0 MiB/2.09 GiB object_store_memory: 0%| | 0/1 [00:00<?, ?it/s]Exception in thread StreamingExecutor-291aea5ad2ee4c7499c11d8ea60b2223:
Traceback (most recent call last):
File "C:\Users\Taurus-Le\AppData\Local\Programs\Python\Python38\lib\threading.py", line 932, in _bootstrap_inner
self.run()
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\streaming_executor.py", line 211, in run
clear_stats_actor_metrics({"dataset": self._dataset_tag})
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\stats.py", line 296, in clear_stats_actor_metrics
_check_cluster_stats_actor()
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\stats.py", line 272, in _check_cluster_stats_actor
current_cluster_id = ray._private.worker._global_node.cluster_id
AttributeError: 'NoneType' object has no attribute 'cluster_id'
Traceback (most recent call last):
File "D:\Work\Python\RayDemo3.8\t.py", line 11, in <module>
preprocessor.fit_transform(ds).to_pandas()
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\dataset.py", line 4247, in to_pandas
count = self.count()
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\dataset.py", line 2501, in count
[get_num_rows.remote(block) for block in self.get_internal_block_refs()]
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\dataset.py", line 4590, in get_internal_block_refs
blocks = self._plan.execute().get_blocks()
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\plan.py", line 599, in execute
blocks = execute_to_legacy_block_list(
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\legacy_compat.py", line 119, in execute_to_legacy_block_list
block_list = _bundles_to_block_list(bundles)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\legacy_compat.py", line 356, in _bundles_to_block_list
for ref_bundle in bundles:
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\interfaces\executor.py", line 37, in __next__
return self.get_next()
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\streaming_executor.py", line 141, in get_next
raise item
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\streaming_executor.py", line 201, in run
while self._scheduling_loop_step(self._topology) and not self._shutdown:
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\streaming_executor.py", line 274, in _scheduling_loop_step
topology[op].dispatch_next_task()
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\streaming_executor_state.py", line 204, in dispatch_next_task
self.op.add_input(inqueue.popleft(), input_index=i)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\interfaces\physical_operator.py", line 276, in add_input
self._add_input_inner(refs, input_index)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\operators\map_operator.py", line 231, in _add_input_inner
self._add_bundled_input(bundle)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\data\_internal\execution\operators\task_pool_map_operator.py", line 64, in _add_bundled_input
gen = map_task.options(**ray_remote_args).remote(
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\remote_function.py", line 245, in remote
return func_cls._remote(args=args, kwargs=kwargs, **updated_options)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\_private\auto_init_hook.py", line 24, in auto_init_wrapper
return fn(*args, **kwargs)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\util\tracing\tracing_helper.py", line 310, in _invocation_remote_span
return method(self, args, kwargs, *_args, **_kwargs)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\remote_function.py", line 267, in _remote
return client_mode_convert_function(self, args, kwargs, **task_options)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\_private\client_mode_hook.py", line 164, in client_mode_convert_function
return client_func._remote(in_args, in_kwargs, **kwargs)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\util\client\common.py", line 298, in _remote
return self.options(**option_args).remote(*args, **kwargs)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\util\client\common.py", line 581, in remote
return return_refs(ray.call_remote(self, *args, **kwargs))
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\util\client\api.py", line 100, in call_remote
return self.worker.call_remote(instance, *args, **kwargs)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\util\client\worker.py", line 562, in call_remote
return self._call_schedule_for_task(task, num_returns)
File "D:\Work\Python\RayDemo3.8\venv\lib\site-packages\ray\util\client\worker.py", line 575, in _call_schedule_for_task
id_futures = [Future() for _ in range(num_return_refs)]
TypeError: 'str' object cannot be interpreted as an integer
What I am supposed to do? Am I missing any important config?