[DataFrame] Fix dtypes #1930

pschafhalter · 2018-04-20T20:47:18Z

What do these changes do?

Calculate dtypes by merging dtypes from all partitions.
Unify dtypes across partitions.
Cache dtypes.

Possible Issues

There are many operations that invalidate the dtypes cache. There's a high chance of edge-case bugs if all operations that modify dataframe data don't invalidate the dtypes cache.

AmplabJenkins · 2018-04-20T21:50:58Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5020/
Test PASSed.

devin-petersohn

There's a lot of complicated logic here, but I think all that is really needed to make sure dtypes are all the same is the following:

1.) Materialize column partitions
2.) Split up column partitions into blocks again
3.) Reassign blocks to new blocks
4.) Dtypes will now be unified because the concat in column partitions will do it for you.

devin-petersohn · 2018-04-20T23:40:43Z

python/ray/dataframe/utils.py

@@ -282,3 +282,60 @@ def decorator(cls):
        return cls

    return decorator
+
+
+def _map(func, *objects):


Why do we need this instead of _map_partitions()?

I thought it would go well with the _reduce() implementation for generic objects in the object store. I can remove it and change back to _map_partitions() instead.

devin-petersohn · 2018-04-22T02:12:15Z

python/ray/dataframe/utils.py

+
+
+@ray.remote
+def _reduce(func, *sequence):


I see what you're doing here, but there might be a cheaper/simpler way.

I'm not sure (at least for getting dtypes) whether this needs a tree-reduce implementation. Regardless, I think you can make it simpler by performing it iteratively, something like

def _reduce(func, oids): while len(oids) > 1: oids = [_remote_reduce_helper.remote(func, item1, item2) for item1, item2 in pairwise(oids)] return oids[0]

You can use _submit to perform on more than just a pair at a time, but this way you construct the entire computation graph locally.

pschafhalter · 2018-04-22T05:22:36Z

@devin-petersohn Thanks for the feedback!

That would work as well, but isn't materializing column partitions and splitting them up back into blocks be more expensive? From my understanding, this would result in 2 expensive operations on partitions:

Convert block partitions to column partitions.
Convert column partitions back to block partitions.

Whereas this implementation only involves 1 expensive operation on partitions:

Set type of block partitions.

devin-petersohn · 2018-04-23T04:12:46Z

@pschafhalter, I think that materializing the columns would be faster because it's not an iterative computation. You can create the block partitions and then return to blocks in the same remote task:

def correct_dtypes(*column_blocks):
    columns = pandas.concat(column_blocks, axis=0, copy=False)
    blocks = # create blocks from columns

pschafhalter · 2018-04-23T05:22:22Z

@devin-petersohn that makes sense. In order to get an object ID for each corrected block, I need to specify num_return_types in correct_dtypes, right?

devin-petersohn · 2018-04-23T06:04:49Z

@pschafhalter Yes, you can do that with correct_dtypes._submit(args=(*args), num_return_vals=n)

p-yang · 2018-04-23T06:43:00Z

python/ray/dataframe/utils.py

+        Object returned by cumulatively applying func.
+    """
+    size = len(sequence)
+    if size == 1:


Generally most tree-reduce implementations don't reduce down to one value due to overhead, so it's probably more worth it to place a larger base case size and enforce that the reduction function is compatible with python's builtin reduce.

p-yang · 2018-04-23T06:43:36Z

python/ray/dataframe/utils.py

+    if size == 1:
+        return sequence[0]
+    left = ray.get(_reduce.remote(func, *sequence[:size // 2]))
+    right = ray.get(_reduce.remote(func, *sequence[size // 2:]))


You should start both remote tasks first, then perform the get afterwards.

pschafhalter · 2018-04-27T06:01:37Z

@devin-petersohn @Veryku Sorry for sitting on this for a while. The latest commit should implement the suggested changes.

AmplabJenkins · 2018-04-27T07:05:00Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5081/
Test PASSed.

AmplabJenkins · 2018-04-27T07:55:30Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5082/
Test PASSed.

kunalgosar · 2018-04-27T08:51:16Z

Brief performance test:

This was done on a file with 79 columns and of size 1.4mb:

In [17]: df = pd.read_csv("/Users/kunalgosar/Downloads/fundamentals.csv")

In [18]: %time df.dtypes
CPU times: user 22.7 ms, sys: 2.97 ms, total: 25.6 ms
Wall time: 103 ms

This was done on a file with 31 columns and of size 150mb:

In [19]: df = pd.read_csv("/Users/kunalgosar/Downloads/creditcard.csv")

In [20]: %time df.dtypes
CPU times: user 21.8 ms, sys: 2.98 ms, total: 24.8 ms
Wall time: 1.47 s

AmplabJenkins · 2018-04-27T09:44:52Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5085/
Test PASSed.

kunalgosar · 2018-04-27T16:32:08Z

python/ray/dataframe/dataframe.py

-                                             self._row_partitions[0]))
-        result.index = self.columns
-        return result
+        # Cache dtypes if necessary


Can this be made as an OID initially, and it is got and cached when called for. This is similar to index.

kunalgosar · 2018-04-27T16:33:20Z

python/ray/dataframe/dataframe.py

@@ -71,6 +72,7 @@ def __init__(self, data=None, index=None, columns=None, dtype=None,
                Metadata for the new dataframe's columns
        """
        self._row_metadata = self._col_metadata = None
+        self._dtypes = None


Once the code below is made to store a remote object, make a dummy call to it here to start the asynchronous call, but do not get the object yet. That is a blocking step and can be done as needed.

pschafhalter · 2018-04-27T19:07:48Z

@kunalgosar most of that cost in the performance test is from materializing the DataFrame through read_csv:

In [3]: df = pd.read_csv("/home/peter/Downloads/creditcard.csv")

In [4]: df_repr = repr(df)

In [5]: df._dtypes # Not cached yet

In [6]: %time df.dtypes
CPU times: user 23.7 ms, sys: 2.23 ms, total: 25.9 ms
Wall time: 231 ms

AmplabJenkins · 2018-04-30T02:50:59Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5121/
Test PASSed.

AmplabJenkins · 2018-05-01T22:31:52Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5134/
Test FAILed.

p-yang

Some functions should have well-known output dtypes (isna comes to mind again). Would it be possible to pass dtypes_cache for those functions?

p-yang · 2018-05-02T00:46:41Z

python/ray/dataframe/dataframe.py

@@ -112,6 +118,10 @@ def __init__(self, data=None, index=None, columns=None, dtype=None,
                elif col_partitions is not None:
                    axis = 1
                    partitions = col_partitions
+                    # All partitions will already have correct dtypes
+                    self._dtypes_cache = [


Is it more performant to have a special case here for col_partitions? Otherwise I'd imagine this is already handled by the added code below.

I think so. Otherwise, we create block partitions from column partitions, materialize column partitions again to correct dtypes, and split that back into block partitions.

pschafhalter · 2018-05-02T22:30:03Z

@Veryku agreed on passing dtypes_cache for function with well-known output types. I discussed this with @devin-petersohn, and created issue #1987. I'll add the optimizations in a subsequent PR.

AmplabJenkins · 2018-05-03T00:33:36Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5152/
Test PASSed.

bug fixes Unify dtypes on DataFrame creation Formatting and comments Cache dtypes Fix bug in _merge_dtypes Fix bug Changed caching logic Fix dtypes issue in read_csv Invalidate dtypes cache when inserting column Simplify unifying dtypes and improve caching Fix typo Better caching of dtypes Fix merge conflicts

AmplabJenkins · 2018-05-03T03:42:02Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/5154/
Test PASSed.

devin-petersohn · 2018-05-03T06:04:02Z

Passes private-travis. Thanks @pschafhalter!

* magic-methods: fmt Fix IndentationError Write magic methods for SampleBatch/PartialRollout Clean up syntax for supported Python versions. (ray-project#1963) [DataFrame] Implements mode, to_datetime, and get_dummies (ray-project#1956) [DataFrame] Fix dtypes (ray-project#1930) keep_dims -> keepdims (ray-project#1980) add pthread linking (ray-project#1986) [DataFrame] Add layer of abstraction to allow OID instantiation (ray-project#1984) [DataFrame] Fix blocking issue on _IndexMetadata passing (ray-project#1965)

* master: (21 commits) Expand local_dir in Trial init (ray-project#2013) Fixing ascii error for Python2 (ray-project#2009) [DataFrame] Implements df.update (ray-project#1997) [DataFrame] Implements df.as_matrix (ray-project#2001) [DataFrame] Implement quantile (ray-project#1992) [DataFrame] Impement sort_values and sort_index (ray-project#1977) [DataFrame] Implement rank (ray-project#1991) [DataFrame] Implemented prod, product, added test suite (ray-project#1994) [DataFrame] Implemented __setitem__, select_dtypes, and astype (ray-project#1941) [DataFrame] Implement diff (ray-project#1996) [DataFrame] Implemented nunique, skew (ray-project#1995) [DataFrame] Implements filter and dropna (ray-project#1959) [DataFrame] Implements df.pipe (ray-project#1999) [DataFrame] Apply() for Lists and Dicts (ray-project#1973) Clean up syntax for supported Python versions. (ray-project#1963) [DataFrame] Implements mode, to_datetime, and get_dummies (ray-project#1956) [DataFrame] Fix dtypes (ray-project#1930) keep_dims -> keepdims (ray-project#1980) add pthread linking (ray-project#1986) [DataFrame] Add layer of abstraction to allow OID instantiation (ray-project#1984) ...

* master: (25 commits) [DataFrame] Add direct pandas imports for MVP (ray-project#1960) Make ActorHandles pickleable, also make proper ActorHandle and ActorC… (ray-project#2007) Expand local_dir in Trial init (ray-project#2013) Fixing ascii error for Python2 (ray-project#2009) [DataFrame] Implements df.update (ray-project#1997) [DataFrame] Implements df.as_matrix (ray-project#2001) [DataFrame] Implement quantile (ray-project#1992) [DataFrame] Impement sort_values and sort_index (ray-project#1977) [DataFrame] Implement rank (ray-project#1991) [DataFrame] Implemented prod, product, added test suite (ray-project#1994) [DataFrame] Implemented __setitem__, select_dtypes, and astype (ray-project#1941) [DataFrame] Implement diff (ray-project#1996) [DataFrame] Implemented nunique, skew (ray-project#1995) [DataFrame] Implements filter and dropna (ray-project#1959) [DataFrame] Implements df.pipe (ray-project#1999) [DataFrame] Apply() for Lists and Dicts (ray-project#1973) Clean up syntax for supported Python versions. (ray-project#1963) [DataFrame] Implements mode, to_datetime, and get_dummies (ray-project#1956) [DataFrame] Fix dtypes (ray-project#1930) keep_dims -> keepdims (ray-project#1980) ...

pschafhalter requested a review from devin-petersohn April 20, 2018 20:47

devin-petersohn reviewed Apr 22, 2018

View reviewed changes

p-yang reviewed Apr 23, 2018

View reviewed changes

pschafhalter force-pushed the df-fix-dtypes branch from 237e963 to a37a0e4 Compare April 27, 2018 06:51

kunalgosar suggested changes Apr 27, 2018

View reviewed changes

pschafhalter force-pushed the df-fix-dtypes branch from a67a931 to 7f6f703 Compare May 1, 2018 22:00

p-yang reviewed May 2, 2018

View reviewed changes

pschafhalter mentioned this pull request May 2, 2018

[DataFrame] Improve performance by passing dtypes_cache to the DataFrame constructor #1987

Closed

pschafhalter force-pushed the df-fix-dtypes branch from 7f6f703 to 27f3929 Compare May 2, 2018 23:25

pschafhalter added 2 commits May 2, 2018 19:36

Correct dtypes on initialization

e715e12

pschafhalter force-pushed the df-fix-dtypes branch from 27f3929 to e715e12 Compare May 3, 2018 02:37

devin-petersohn approved these changes May 3, 2018

View reviewed changes

devin-petersohn merged commit d67b786 into ray-project:master May 3, 2018

pschafhalter deleted the df-fix-dtypes branch May 22, 2018 17:03

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[DataFrame] Fix dtypes #1930

[DataFrame] Fix dtypes #1930

pschafhalter commented Apr 20, 2018

AmplabJenkins commented Apr 20, 2018

devin-petersohn left a comment

devin-petersohn Apr 20, 2018

pschafhalter Apr 22, 2018

devin-petersohn Apr 22, 2018

p-yang Apr 23, 2018

pschafhalter commented Apr 22, 2018

devin-petersohn commented Apr 23, 2018

pschafhalter commented Apr 23, 2018

devin-petersohn commented Apr 23, 2018 •

edited

Loading

p-yang Apr 23, 2018

p-yang Apr 23, 2018

pschafhalter commented Apr 27, 2018

AmplabJenkins commented Apr 27, 2018

AmplabJenkins commented Apr 27, 2018

kunalgosar commented Apr 27, 2018

AmplabJenkins commented Apr 27, 2018

kunalgosar Apr 27, 2018

kunalgosar Apr 27, 2018

pschafhalter commented Apr 27, 2018

AmplabJenkins commented Apr 30, 2018

AmplabJenkins commented May 1, 2018

p-yang left a comment

p-yang May 2, 2018

pschafhalter May 2, 2018

pschafhalter commented May 2, 2018

AmplabJenkins commented May 3, 2018

AmplabJenkins commented May 3, 2018

devin-petersohn commented May 3, 2018

[DataFrame] Fix dtypes #1930

[DataFrame] Fix dtypes #1930

Conversation

pschafhalter commented Apr 20, 2018

What do these changes do?

Possible Issues

AmplabJenkins commented Apr 20, 2018

devin-petersohn left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pschafhalter commented Apr 22, 2018

devin-petersohn commented Apr 23, 2018

pschafhalter commented Apr 23, 2018

devin-petersohn commented Apr 23, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pschafhalter commented Apr 27, 2018

AmplabJenkins commented Apr 27, 2018

AmplabJenkins commented Apr 27, 2018

kunalgosar commented Apr 27, 2018

AmplabJenkins commented Apr 27, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pschafhalter commented Apr 27, 2018

AmplabJenkins commented Apr 30, 2018

AmplabJenkins commented May 1, 2018

p-yang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pschafhalter commented May 2, 2018

AmplabJenkins commented May 3, 2018

AmplabJenkins commented May 3, 2018

devin-petersohn commented May 3, 2018

devin-petersohn commented Apr 23, 2018 •

edited

Loading