mars-project / mars-project/mars

[BUG] read_csv().apply(func, axis=1) failed

Open
#1,107 1 comment 0 reactions 0 assignees View on GitHub
mod: dataframe type: bug
Dominant language
Python
Stars
2.7k
Forks
325
PR merge metrics
No merged PRs in 30d

Description

**Describe the bug**

read_csv().apply(func, axis=1) failed.

**To Reproduce**
To help us reproducing this bug, please provide information below:
1. Your Python version
2. The version of Mars you use
3. Versions of crucial packages, such as numpy, scipy and protobuf
4. Full stack of the error.
5. Minimized code to reproduce the error.

```python
In [44]: %%time
...: df = md.read_csv('Downloads/world-cities-database/worldcitiespop.csv', chunk_bytes='16M')
...: df.apply(lambda r: haversine(r['Latitude'], r['Longitude'], 30.25, 120.17), result_type='reduce', axis=1).execute
...: ()
...:
...:
---------------------------------------------------------------------------
ValueError Traceback (most recent call last)
in

~/Workspace/mars/mars/core.py in execute(self, session, **kw)
426 if session is None:
427 session = Session.default_or_local()
--> 428 return session.run(self, **kw)
429
430 def fetch(self, session=None, **kw):

~/Workspace/mars/mars/session.py in run(self, *tileables, **kw)
181 tileables = tuple(mt.tensor(t) if not isinstance(t, (Entity, Base)) else t
182 for t in tileables)
--> 183 result = self._sess.run(*tileables, **kw)
184
185 for t in tileables:

~/Workspace/mars/mars/session.py in run(self, *tileables, **kw)
88 # set number of running cores
89 self.context.set_ncores(kw['n_parallel'])
---> 90 res = self._executor.execute_tileables(tileables, **kw)
91 return res
92

~/Workspace/mars/mars/utils.py in _wrapped(*args, **kwargs)
380 _kernel_mode.eager = False
381 _kernel_mode.eager_count = enter_eager_count + 1
--> 382 return func(*args, **kwargs)
383 finally:
384 _kernel_mode.eager_count -= 1

~/Workspace/mars/mars/utils.py in inner(*args, **kwargs)
468 def inner(*args, **kwargs):
469 with build_mode():
--> 470 return func(*args, **kwargs)
471 return inner
472

~/Workspace/mars/mars/executor.py in execute_tileables(self, tileables, fetch, n_parallel, n_thread, print_progress, mock, compose)
810 # build chunk graph, tile will be done during building
811 chunk_graph = chunk_graph_builder.build(
--> 812 tileables, tileable_graph=tileable_graph)
813 tileable_graph = chunk_graph_builder.prev_tileable_graph
814 temp_result_keys = set(result_keys)

~/Workspace/mars/mars/utils.py in _wrapped(*args, **kwargs)
380 _kernel_mode.eager = False
381 _kernel_mode.eager_count = enter_eager_count + 1
--> 382 return func(*args, **kwargs)
383 finally:
384 _kernel_mode.eager_count -= 1

~/Workspace/mars/mars/utils.py in inner(*args, **kwargs)
468 def inner(*args, **kwargs):
469 with build_mode():
--> 470 return func(*args, **kwargs)
471 return inner
472

~/Workspace/mars/mars/tiles.py in build(self, tileables, tileable_graph)
338
339 chunk_graph = super().build(
--> 340 tileables, tileable_graph=tileable_graph)
341 self._iterative_chunk_graphs.append(chunk_graph)
342 if len(self._interrupted_ops) == 0:

~/Workspace/mars/mars/utils.py in _wrapped(*args, **kwargs)
380 _kernel_mode.eager = False
381 _kernel_mode.eager_count = enter_eager_count + 1
--> 382 return func(*args, **kwargs)
383 finally:
384 _kernel_mode.eager_count -= 1

~/Workspace/mars/mars/utils.py in inner(*args, **kwargs)
468 def inner(*args, **kwargs):
469 with build_mode():
--> 470 return func(*args, **kwargs)
471 return inner
472

~/Workspace/mars/mars/tiles.py in build(self, tileables, tileable_graph)
252 # for further execution
253 partial_tiled_chunks = \
--> 254 self._on_tile_failure(tileable_data.op, exc_info)
255 if partial_tiled_chunks is not None and \
256 len(partial_tiled_chunks) > 0:

~/Workspace/mars/mars/tiles.py in inner(op, exc_info)
290 on_tile_failure(op, exc_info)
291 else:
--> 292 raise exc_info[1].with_traceback(exc_info[2]) from None
293 return inner
294

~/Workspace/mars/mars/tiles.py in build(self, tileables, tileable_graph)
232 continue
233 try:
--> 234 tiled = self._tile(tileable_data, tileable_graph)
235 tiled_op.add(tileable_data.op)
236 for t, td in zip(tileable_data.op.outputs, tiled):

~/Workspace/mars/mars/tiles.py in _tile(self, tileable_data, tileable_graph)
326 if any(inp.op in self._interrupted_ops for inp in tileable_data.inputs):
327 raise TilesError('Tile fail due to failure of inputs')
--> 328 return super()._tile(tileable_data, tileable_graph)
329
330 @kernel_mode

~/Workspace/mars/mars/tiles.py in _tile(self, tileable_data, tileable_graph)
190 t._nsplits = o.nsplits
191 elif on_tile is None:
--> 192 tds[0]._inplace_tile()
193 else:
194 tds = on_tile(tileable_data.op.outputs, tds)

~/Workspace/mars/mars/core.py in _inplace_tile(self)
160
161 def _inplace_tile(self):
--> 162 return handler.inplace_tile(self)
163
164 def __getattr__(self, attr):

~/Workspace/mars/mars/tiles.py in inplace_tile(self, to_tile)
125 if not to_tile.is_coarse():
126 return to_tile
--> 127 dispatched = self.dispatch(to_tile.op)
128 self._assign_to([d.data for d in dispatched], to_tile.op.outputs)
129 return to_tile

~/Workspace/mars/mars/utils.py in _wrapped(*args, **kwargs)
380 _kernel_mode.eager = False
381 _kernel_mode.eager_count = enter_eager_count + 1
--> 382 return func(*args, **kwargs)
383 finally:
384 _kernel_mode.eager_count -= 1

~/Workspace/mars/mars/tiles.py in dispatch(self, op)
113 return self._handlers[op_cls](op)
114 try:
--> 115 return op_cls.tile(op)
116 except NotImplementedError:
117 for registered_op_cls in self._handlers:

~/Workspace/mars/mars/dataframe/base/apply.py in tile(cls, op)
161 def tile(cls, op):
162 if op.inputs[0].op.object_type == ObjectType.dataframe:
--> 163 return cls._tile_df(op)
164 else:
165 return cls._tile_series(op)

~/Workspace/mars/mars/dataframe/base/apply.py in _tile_df(cls, op)
101 if axis == 1:
102 chunk_size = chunk_size[::-1]
--> 103 in_df = in_df.rechunk(chunk_size).tiles()
104
105 chunks = []

~/Workspace/mars/mars/dataframe/base/rechunk.py in rechunk(a, chunk_size, threshold, chunk_size_limit)
88 else:
89 itemsize = a.dtype.itemsize
---> 90 chunk_size = get_nsplits(a, chunk_size, itemsize)
91 if chunk_size == a.nsplits:
92 return a

~/Workspace/mars/mars/tensor/rechunk/core.py in get_nsplits(tileable, new_chunk_size, itemsize)
36 chunk_size = new_chunk_size
37
---> 38 return decide_chunk_sizes(tileable.shape, chunk_size, itemsize)
39
40

~/Workspace/mars/mars/tensor/utils.py in decide_chunk_sizes(shape, chunk_size, itemsize)
560 raise ValueError("chunks have more dimensions than input tensor")
561 if nleft == 0:
--> 562 return normalize_chunk_sizes(shape, tuple(chunk_size[j] for j in range(len(shape))))
563
564 max_chunk_size = options.chunk_store_limit

~/Workspace/mars/mars/tensor/utils.py in normalize_chunk_sizes(shape, chunk_size)
69 assert isinstance(chunk, int)
70
---> 71 sizes = tuple(chunk for _ in range(int(size / chunk))) + \
72 (tuple() if size % chunk == 0 else (size % chunk,))
73 chunk_sizes.append(sizes)

ValueError: cannot convert float NaN to integer
```

Contributor guide

Open the contributing guide

Research direction

Start with the failing example in the issue and trace the stack through mars/dataframe/base/apply.py, mars/dataframe/base/rechunk.py, and mars/tensor/utils.py. Reproduce the ValueError from read_csv().apply(..., axis=1), then verify the reported operation completes successfully without the NaN conversion failure.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.