[Dataframe] Change pandas and ray.dataframe imports

python/ray/dataframe/__init__.py

@@ -2,7 +2,7 @@
 from __future__ import division
 from __future__ import print_function
 
-import pandas as pd
+import pandas
 # TODO: In the future `set_option` or similar needs to run on every node
 # in order to keep all pandas instances across nodes consistent
 from pandas import (eval, unique, value_counts, cut, to_numeric, factorize,
@@ -12,11 +12,11 @@
                     set_option, NaT, PeriodIndex, Categorical)
 import threading
 
-pd_version = pd.__version__
-pd_major = int(pd_version.split(".")[0])
-pd_minor = int(pd_version.split(".")[1])
+pandas_version = pandas.__version__
+pandas_major = int(pandas_version.split(".")[0])
+pandas_minor = int(pandas_version.split(".")[1])
 
-if pd_major == 0 and pd_minor != 22:
+if pandas_major == 0 and pandas_minor != 22:
     raise Exception("In order to use Pandas on Ray, your pandas version must "
                     "be 0.22. You can run 'pip install pandas==0.22'")
 

python/ray/dataframe/groupby.py

@@ -2,7 +2,7 @@
 from __future__ import division
 from __future__ import print_function
 
-import pandas as pd
+import pandas
 import numpy as np
 import pandas.core.groupby
 from pandas.core.dtypes.common import is_list_like
@@ -12,6 +12,7 @@
 
 from .utils import _inherit_docstrings, _reindex_helper
 from .concat import concat
+from .index_metadata import _IndexMetadata
 
 
 @_inherit_docstrings(pandas.core.groupby.DataFrameGroupBy,
@@ -31,12 +32,13 @@ def __init__(self, df, by, axis, level, as_index, sort, group_keys,
 
         if axis == 0:
             partitions = [column for column in df._block_partitions.T]
-            self._index_grouped = pd.Series(self._index, index=self._index)\
+            self._index_grouped = \
+                pandas.Series(self._index, index=self._index) \
                 .groupby(by=by, sort=sort)
         else:
             partitions = [row for row in df._block_partitions]
             self._index_grouped = \
-                pd.Series(self._columns, index=self._columns) \
+                pandas.Series(self._columns, index=self._columns) \
                 .groupby(by=by, sort=sort)
 
         self._keys_and_values = [(k, v)
@@ -127,7 +129,7 @@ def tshift(self):
 
     @property
     def groups(self):
-        return {k: pd.Index(v) for k, v in self._keys_and_values}
+        return {k: pandas.Index(v) for k, v in self._keys_and_values}
 
     def min(self, **kwargs):
         return self._apply_agg_function(lambda df: df.min(axis=self._axis,
@@ -194,7 +196,7 @@ def apply_helper(df):
 
         result = [func(v) for k, v in self._iter]
         if self._axis == 0:
-            if isinstance(result[0], pd.Series):
+            if isinstance(result[0], pandas.Series):
                 # Applied an aggregation function
                 new_df = concat(result, axis=1).T
                 new_df.columns = self._columns
@@ -208,8 +210,11 @@ def apply_helper(df):
                     num_return_vals=len(new_df._block_partitions))
                     for block in new_df._block_partitions.T]).T
                 new_df.index = self._index
+                new_df._row_metadata = \
+                    _IndexMetadata(new_df._block_partitions[:, 0],
+                                   index=new_df.index, axis=0)
         else:
-            if isinstance(result[0], pd.Series):
+            if isinstance(result[0], pandas.Series):
                 # Applied an aggregation function
                 new_df = concat(result, axis=1)
                 new_df.columns = [k for k, v in self._iter]
@@ -223,6 +228,9 @@ def apply_helper(df):
                     num_return_vals=new_df._block_partitions.shape[1])
                     for block in new_df._block_partitions])
                 new_df.columns = self._columns
+                new_df._col_metadata = \
+                    _IndexMetadata(new_df._block_partitions[0, :],
+                                   index=new_df.columns, axis=1)
         return new_df
 
     @property
@@ -392,6 +400,9 @@ def head(self, n=5):
                 num_return_vals=len(new_df._block_partitions))
                 for block in new_df._block_partitions.T]).T
             new_df.index = sorted_index
+            new_df._row_metadata = \
+                _IndexMetadata(new_df._block_partitions[:, 0],
+                               index=new_df.index, axis=0)
 
         return new_df
 
@@ -447,6 +458,9 @@ def tail(self, n=5):
                 num_return_vals=len(new_df._block_partitions))
                 for block in new_df._block_partitions.T]).T
             new_df.index = sorted_index
+            new_df._row_metadata = \
+                _IndexMetadata(new_df._block_partitions[:, 0],
+                               index=new_df.index, axis=0)
 
         return new_df
 
@@ -516,6 +530,9 @@ def _apply_df_function(self, f, concat_axis=None):
                 num_return_vals=len(new_df._block_partitions))
                 for block in new_df._block_partitions.T]).T
             new_df.index = self._index
+            new_df._row_metadata = \
+                _IndexMetadata(new_df._block_partitions[:, 0],
+                               index=new_df.index, axis=0)
         else:
             new_df._block_partitions = np.array([_reindex_helper._submit(
                 args=tuple([new_df.columns, self._columns, 0,
@@ -524,14 +541,17 @@ def _apply_df_function(self, f, concat_axis=None):
                 num_return_vals=new_df._block_partitions.shape[1])
                 for block in new_df._block_partitions])
             new_df.columns = self._columns
+            new_df._col_metadata = \
+                _IndexMetadata(new_df._block_partitions[0, :],
+                               index=new_df.columns, axis=1)
 
         return new_df
 
 
 @ray.remote
 def groupby(by, axis, level, as_index, sort, group_keys, squeeze, *df):
 
-    df = pd.concat(df, axis=axis)
+    df = pandas.concat(df, axis=axis)
 
     return [v for k, v in df.groupby(by=by,
                                      axis=axis,

python/ray/dataframe/index_metadata.py

@@ -1,11 +1,12 @@
-import pandas as pd
+import pandas
 import numpy as np
 import ray
 
 from .utils import (
     _build_row_lengths,
     _build_col_widths,
-    _build_coord_df)
+    _build_coord_df,
+    _check_empty)
 
 from pandas.core.indexing import convert_to_index_sliceable
 
@@ -32,12 +33,12 @@ def __init__(self, dfs=None, index=None, axis=0, lengths_oid=None,
 
         Args:
             dfs ([ObjectID]): ObjectIDs of dataframe partitions
-            index (pd.Index): Index of the Ray DataFrame.
+            index (pandas.Index): Index of the Ray DataFrame.
             axis: Axis of partition (0=row partitions, 1=column partitions)
 
         Returns:
-            A IndexMetadata backed by the specified pd.Index, partitioned off
-            specified partitions
+            A IndexMetadata backed by the specified pandas.Index, partitioned
+            off specified partitions
         """
         assert (lengths_oid is None) == (coord_df_oid is None), \
             "Must pass both or neither of lengths_oid and coord_df_oid"
@@ -48,6 +49,9 @@ def __init__(self, dfs=None, index=None, axis=0, lengths_oid=None,
             else:
                 lengths_oid = _build_col_widths.remote(dfs)
             coord_df_oid = _build_coord_df.remote(lengths_oid, index)
+            self._empty = _check_empty.remote(dfs)
+        else:
+            self._empty = True
 
         self._lengths = lengths_oid
         self._coord_df = coord_df_oid
@@ -115,7 +119,7 @@ def _set_index(self, new_index):
         This design is more straightforward than caching indexes on setting the
         coord_df to an OID due to the possibility of an OID-to-OID change.
         """
-        new_index = pd.DataFrame(index=new_index).index
+        new_index = pandas.DataFrame(index=new_index).index
         assert len(new_index) == len(self)
 
         self._index_cache = new_index
@@ -138,7 +142,7 @@ def _get_index_cache(self):
             The Index object in _index_cache.
         """
         if self._index_cache_validator is None:
-            self._index_cache_validator = pd.RangeIndex(len(self))
+            self._index_cache_validator = pandas.RangeIndex(len(self))
         elif isinstance(self._index_cache_validator,
                         ray.ObjectID):
             self._index_cache_validator = ray.get(self._index_cache_validator)
@@ -157,6 +161,16 @@ def _set_index_cache(self, new_index):
     # cache to accept ObjectIDs and ray.get them when needed.
     _index_cache = property(_get_index_cache, _set_index_cache)
 
+    def _get_empty(self):
+        if isinstance(self._empty_cache, ray.ObjectID):
+            self._empty_cache = ray.get(self._empty_cache)
+        return self._empty_cache
+
+    def _set_empty(self, empty):
+        self._empty_cache = empty
+
+    _empty = property(_get_empty, _set_empty)
+
     def coords_of(self, key):
         """Returns the coordinates (partition, index_within_partition) of the
         provided key in the index. Can be called on its own or implicitly
@@ -170,9 +184,9 @@ def coords_of(self, key):
 
         Returns:
             Pandas object with the keys specified. If key is a single object
-            it will be a pd.Series with items `partition` and
+            it will be a pandas.Series with items `partition` and
             `index_within_partition`, and if key is a slice or if the key is
-            duplicate it will be a pd.DataFrame with said items as columns.
+            duplicate it will be a pandas.DataFrame with said items as columns.
         """
         return self._coord_df.loc[key]
 
@@ -191,7 +205,7 @@ def partition_series(self, partition):
                     'index_within_partition']
 
     def __len__(self):
-        return sum(self._lengths)
+        return int(sum(self._lengths))
 
     def reset_partition_coords(self, partitions=None):
         partitions = np.array(partitions)
@@ -200,7 +214,7 @@ def reset_partition_coords(self, partitions=None):
             partition_mask = (self._coord_df['partition'] == partition)
             # Since we are replacing columns with RangeIndex inside the
             # partition, we have to make sure that our reference to it is
-            # updated as well.
+            # upandasated as well.
             try:
                 self._coord_df.loc[partition_mask,
                                    'index_within_partition'] = np.arange(
@@ -263,7 +277,7 @@ def insert(self, key, loc=None, partition=None,
         # TODO: Determine if there's a better way to do a row-index insert in
         # pandas, because this is very annoying/unsure of efficiency
         # Create new coord entry to insert
-        coord_to_insert = pd.DataFrame(
+        coord_to_insert = pandas.DataFrame(
                 {'partition': partition,
                  'index_within_partition': index_within_partition},
                 index=[key])
@@ -329,9 +343,9 @@ def __getitem__(self, key):
 
         Returns:
             Pandas object with the keys specified. If key is a single object
-            it will be a pd.Series with items `partition` and
+            it will be a pandas.Series with items `partition` and
             `index_within_partition`, and if key is a slice or if the key is
-            duplicate it will be a pd.DataFrame with said items as columns.
+            duplicate it will be a pandas.DataFrame with said items as columns.
         """
         return self.coords_of(key)
 
@@ -355,26 +369,37 @@ def drop(self, labels, errors='raise'):
         """
         dropped = self.coords_of(labels)
 
-        # Update first lengths to prevent possible length inconsistencies
-        if isinstance(dropped, pd.DataFrame):
+        # Upandasate first lengths to prevent possible length inconsistencies
+        if isinstance(dropped, pandas.DataFrame):
             try:
                 drop_per_part = dropped.groupby(["partition"]).size()\
-                        .reindex(index=pd.RangeIndex(len(self._lengths)),
+                        .reindex(index=pandas.RangeIndex(len(self._lengths)),
                                  fill_value=0)
             except ValueError:
                 # Copy the arrow sealed dataframe so we can mutate it.
                 dropped = dropped.copy()
                 drop_per_part = dropped.groupby(["partition"]).size()\
-                    .reindex(index=pd.RangeIndex(len(self._lengths)),
+                    .reindex(index=pandas.RangeIndex(len(self._lengths)),
                              fill_value=0)
-        elif isinstance(dropped, pd.Series):
+        elif isinstance(dropped, pandas.Series):
             drop_per_part = np.zeros_like(self._lengths)
             drop_per_part[dropped["partition"]] = 1
         else:
             raise AssertionError("Unrecognized result from `coords_of`")
-        self._lengths = self._lengths - drop_per_part
 
-        self._coord_df = self._coord_df.drop(labels, errors=errors)
+        self._lengths = self._lengths - np.array(drop_per_part)
+
+        new_coord_df = self._coord_df.drop(labels, errors=errors)
+
+        num_dropped = 0
+        for i, length in enumerate(self._lengths):
+            if length == 0:
+                num_dropped += 1
+            if num_dropped > 0:
+                new_coord_df['partition'][new_coord_df['partition'] == i] \
+                    -= num_dropped
+
+        self._coord_df = new_coord_df
         return dropped
 
     def rename_index(self, mapper):

python/ray/dataframe/indexing.py

@@ -13,7 +13,7 @@
 An illustration is available at
 https://github.com/ray-project/ray/pull/1955#issuecomment-386781826
 """
-import pandas as pd
+import pandas
 import numpy as np
 import ray
 from warnings import warn
@@ -96,7 +96,7 @@ def _is_enlargement(locator, coord_df):
     """
     if is_list_like(locator) and not is_slice(
             locator) and len(locator) > 0 and not is_boolean_array(locator):
-        n_diff_elems = len(pd.Index(locator).difference(coord_df.index))
+        n_diff_elems = len(pandas.Index(locator).difference(coord_df.index))
         is_enlargement_boolean = n_diff_elems > 0
         return is_enlargement_boolean
     return False
@@ -140,8 +140,8 @@ def __init__(self, ray_df):
     def __getitem__(self, row_lookup, col_lookup, ndim):
         """
         Args:
-            row_lookup: A pd dataframe, a partial view from row_coord_df
-            col_lookup: A pd dataframe, a partial view from col_coord_df
+            row_lookup: A pandas dataframe, a partial view from row_coord_df
+            col_lookup: A pandas dataframe, a partial view from col_coord_df
             ndim: the dimension of returned data
         """
         if ndim == 2:
@@ -152,7 +152,7 @@ def __getitem__(self, row_lookup, col_lookup, ndim):
             result = ray.get(_blocks_to_col.remote(*extracted)).squeeze()
 
             if is_scalar(result):
-                result = pd.Series(result)
+                result = pandas.Series(result)
 
             scaler_axis = row_lookup if len(row_lookup) == 1 else col_lookup
             series_name = scaler_axis.iloc[0].name
@@ -213,8 +213,8 @@ def _generate_view(self, row_lookup, col_lookup):
     def __setitem__(self, row_lookup, col_lookup, item):
         """
         Args:
-            row_lookup: A pd dataframe, a partial view from row_coord_df
-            col_lookup: A pd dataframe, a partial view from col_coord_df
+            row_lookup: A pandas dataframe, a partial view from row_coord_df
+            col_lookup: A pandas dataframe, a partial view from col_coord_df
             item: The new item needs to be set. It can be any shape that's
                 broadcastable to the product of the lookup tables.
         """
@@ -348,14 +348,14 @@ def _enlarge_axis(self, locator, axis):
             [self.block_oids, nan_blks], axis=0 if row_based_bool else 1)
 
         # 3. Prepare metadata to return
-        nan_coord_df = pd.DataFrame(data=[{
+        nan_coord_df = pandas.DataFrame(data=[{
             '': name,
             'partition': blk_part_n_row if row_based_bool else blk_part_n_col,
             'index_within_partition': i
         } for name, i in zip(nan_labels, np.arange(num_nan_labels))
         ]).set_index('')
 
-        coord_df = pd.concat([major_meta._coord_df, nan_coord_df])
+        coord_df = pandas.concat([major_meta._coord_df, nan_coord_df])
         coord_df = coord_df.loc[locator]  # Re-index that allows duplicates
 
         lens = major_meta._lengths
@@ -370,7 +370,7 @@ def _compute_enlarge_labels(self, locator, base_index):
         Returns:
              nan_labels: The labels needs to be added
         """
-        locator_as_index = pd.Index(locator)
+        locator_as_index = pandas.Index(locator)
 
         nan_labels = locator_as_index.difference(base_index)
         common_labels = locator_as_index.intersection(base_index)