from __future__ import absolute_import

from __future__ import division

from __future__ import print_function

import pandas as pd

from pandas.api.types import is_scalar

from pandas.util._validators import validate_bool_kwarg

from pandas.core.index import _ensure_index_from_sequences

from pandas._libs import lib

from pandas.core.dtypes.cast import maybe_upcast_putmask

from pandas.compat import lzip

from pandas.core.dtypes.common import (

is_bool_dtype,

is_numeric_dtype,

is_timedelta64_dtype)

import warnings

import numpy as np

import ray

import itertools

from .utils import (

_get_lengths,

to_pandas,

_shuffle,

_local_groupby,

_deploy_func,

_compute_length_and_index,

_prepend_partitions)

class DataFrame(object):

def __init__(self, df, columns, index=None):

"""Distributed DataFrame object backed by Pandas dataframes.

Args:

df ([ObjectID]): The list of ObjectIDs that contain the dataframe

partitions.

columns (pandas.Index): The column names for this dataframe, in

pandas Index object.

index (pandas.Index or list): The row index for this dataframe.

"""

assert(len(df) > 0)

self._df = df

self.columns = columns

# this _index object is a pd.DataFrame

# and we use that DataFrame's Index to index the rows.

self._lengths, self._index = _compute_length_and_index.remote(self._df)

if index is not None:

self.index = index

def __str__(self):

return repr(self)

def __repr__(self):

if sum(self._lengths) < 40:

result = repr(to_pandas(self))

return result

head = repr(to_pandas(self.head(20)))

tail = repr(to_pandas(self.tail(20)))

result = head + "\n...\n" + tail

return result

def _get_index(self):

"""Get the index for this DataFrame.

Returns:

The union of all indexes across the partitions.

"""

return self._index.index

def _set_index(self, new_index):

"""Set the index for this DataFrame.

Args:

new_index: The new index to set this

"""

self._index.index = new_index

index = property(_get_index, _set_index)

def _get__index(self):

"""Get the _index for this DataFrame.

Returns:

The default index.

"""

if isinstance(self._index_cache, ray.local_scheduler.ObjectID):

self._index_cache = ray.get(self._index_cache)

return self._index_cache

def _set__index(self, new__index):

"""Set the _index for this DataFrame.

Args:

new__index: The new default index to set.

"""

self._index_cache = new__index

_index = property(_get__index, _set__index)

def _compute_lengths(self):

"""Updates the stored lengths of DataFrame partions

"""

self._lengths = [_deploy_func.remote(_get_lengths, d)

for d in self._df]

def _get_lengths(self):

"""Gets the lengths for each partition and caches it if it wasn't.

Returns:

A list of integers representing the length of each partition.

"""

if isinstance(self._length_cache, ray.local_scheduler.ObjectID):

self._length_cache = ray.get(self._length_cache)

elif isinstance(self._length_cache, list) and \

isinstance(self._length_cache[0],

ray.local_scheduler.ObjectID):

self._length_cache = ray.get(self._length_cache)

return self._length_cache

def _set_lengths(self, lengths):

"""Sets the lengths of each partition for this DataFrame.

We use this because we can compute it when creating the DataFrame.

Args:

lengths ([ObjectID or Int]): A list of lengths for each

partition, in order.

"""

self._length_cache = lengths

_lengths = property(_get_lengths, _set_lengths)

@property

def size(self):

"""Get the number of elements in the DataFrame.

Returns:

The number of elements in the DataFrame.

"""

return len(self.index) * len(self.columns)

@property

def ndim(self):

"""Get the number of dimensions for this DataFrame.

Returns:

The number of dimensions for this DataFrame.

"""

# The number of dimensions is common across all partitions.

# The first partition will be enough.

return ray.get(_deploy_func.remote(lambda df: df.ndim, self._df[0]))

@property

def ftypes(self):

"""Get the ftypes for this DataFrame.

Returns:

The ftypes for this DataFrame.

"""

# The ftypes are common across all partitions.

# The first partition will be enough.

return ray.get(_deploy_func.remote(lambda df: df.ftypes, self._df[0]))

@property

def dtypes(self):

"""Get the dtypes for this DataFrame.

Returns:

The dtypes for this DataFrame.

"""

# The dtypes are common across all partitions.

# The first partition will be enough.

return ray.get(_deploy_func.remote(lambda df: df.dtypes, self._df[0]))

@property

def empty(self):

"""Determines if the DataFrame is empty.

Returns:

True if the DataFrame is empty.

False otherwise.

"""

all_empty = ray.get(self._map_partitions(lambda df: df.empty)._df)

return False not in all_empty

@property

def values(self):

"""Create a numpy array with the values from this DataFrame.

Returns:

The numpy representation of this DataFrame.

"""

return np.concatenate(

ray.get(self._map_partitions(lambda df: df.values)._df))

@property

def axes(self):

"""Get the axes for the DataFrame.

Returns:

The axes for the DataFrame.

"""

return [self.index, self.columns]

@property

def shape(self):

"""Get the size of each of the dimensions in the DataFrame.

Returns:

A tuple with the size of each dimension as they appear in axes().

"""

return (len(self.index), len(self.columns))

def _map_partitions(self, func, index=None):

"""Apply a function on each partition.

Args:

func (callable): The function to Apply.

Returns:

A new DataFrame containing the result of the function.

"""

assert(callable(func))

new_df = [_deploy_func.remote(func, part) for part in self._df]

if index is None:

index = self.index

return DataFrame(new_df, self.columns, index=index)

def _update_inplace(self, df=None, columns=None, index=None):

"""Updates the current DataFrame inplace

"""

assert(len(df) > 0)

if df is not None:

self._df = df

if columns is not None:

self.columns = columns

self._lengths, self._index = _compute_length_and_index.remote(self._df)

if index is not None:

self.index = index

def add_prefix(self, prefix):

"""Add a prefix to each of the column names.

Returns:

A new DataFrame containing the new column names.

"""

new_cols = self.columns.map(lambda x: str(prefix) + str(x))

return DataFrame(self._df, new_cols, index=self.index)

def add_suffix(self, suffix):

"""Add a suffix to each of the column names.

Returns:

A new DataFrame containing the new column names.

"""

new_cols = self.columns.map(lambda x: str(x) + str(suffix))

return DataFrame(self._df, new_cols, index=self.index)

def applymap(self, func):

"""Apply a function to a DataFrame elementwise.

Args:

func (callable): The function to apply.

"""

assert(callable(func))

return self._map_partitions(lambda df: df.applymap(lambda x: func(x)))

def copy(self, deep=True):

"""Creates a shallow copy of the DataFrame.

Returns:

A new DataFrame pointing to the same partitions as this one.

"""

return DataFrame(self._df, self.columns, index=self.index)

def groupby(self, by=None, axis=0, level=None, as_index=True, sort=True,

group_keys=True, squeeze=False, **kwargs):

"""Apply a groupby to this DataFrame. See _groupby() remote task.

Args:

by: The value to groupby.

axis: The axis to groupby.

level: The level of the groupby.

as_index: Whether or not to store result as index.

group_keys: Whether or not to group the keys.

squeeze: Whether or not to squeeze.

Returns:

A new DataFrame resulting from the groupby.

"""

indices = self.index.unique()

chunksize = int(len(indices) / len(self._df))

partitions = [_shuffle.remote(df, indices, chunksize)

for df in self._df]

partitions = ray.get(partitions)

# Transpose the list of dataframes

# TODO find a better way

shuffle = []

for i in range(len(partitions[0])):

shuffle.append([])

for j in range(len(partitions)):

shuffle[i].append(partitions[j][i])

new_dfs = [_local_groupby.remote(part, axis=axis) for part in shuffle]

return DataFrame(new_dfs, self.columns, index=indices)

def reduce_by_index(self, func, axis=0):

"""Perform a reduction based on the row index.

Args:

func (callable): The function to call on the partition

after the groupby.

Returns:

A new DataFrame with the result of the reduction.

"""

return self.groupby(axis=axis)._map_partitions(

func, index=pd.unique(self.index))

def sum(self, axis=None, skipna=True, level=None, numeric_only=None):

"""Perform a sum across the DataFrame.

Args:

axis (int): The axis to sum on.

skipna (bool): True to skip NA values, false otherwise.

Returns:

The sum of the DataFrame.

"""

intermediate_index = [idx

for _ in range(len(self._df))

for idx in self.columns]

sum_of_partitions = self._map_partitions(

lambda df: df.sum(axis=axis, skipna=skipna, level=level,

numeric_only=numeric_only),

index=intermediate_index)

return sum_of_partitions.reduce_by_index(

lambda df: df.sum(axis=axis, skipna=skipna, level=level,

numeric_only=numeric_only))

def abs(self):

"""Apply an absolute value function to all numberic columns.

Returns:

A new DataFrame with the applied absolute value.

"""

for t in self.dtypes:

if np.dtype('O') == t:

# TODO Give a more accurate error to Pandas

raise TypeError("bad operand type for abs():", "str")

return self._map_partitions(lambda df: df.abs())

def isin(self, values):

"""Fill a DataFrame with booleans for cells contained in values.

Args:

values (iterable, DataFrame, Series, or dict): The values to find.

Returns:

A new DataFrame with booleans representing whether or not a cell

is in values.

True: cell is contained in values.

False: otherwise

"""

return self._map_partitions(lambda df: df.isin(values))

def isna(self):

"""Fill a DataFrame with booleans for cells containing NA.

Returns:

A new DataFrame with booleans representing whether or not a cell

is NA.

True: cell contains NA.

False: otherwise.

"""

return self._map_partitions(lambda df: df.isna())

def isnull(self):

"""Fill a DataFrame with booleans for cells containing a null value.

Returns:

A new DataFrame with booleans representing whether or not a cell

is null.

True: cell contains null.

False: otherwise.

"""

return self._map_partitions(lambda df: df.isnull)

def keys(self):

"""Get the info axis for the DataFrame.

Returns:

A pandas Index for this DataFrame.

"""

# Each partition should have the same index, so we'll use 0's

return self.columns

def transpose(self, *args, **kwargs):

"""Transpose columns and rows for the DataFrame.

Note: Triggers a shuffle.

Returns:

A new DataFrame transposed from this DataFrame.

"""

temp_index = [idx

for _ in range(len(self._df))

for idx in self.columns]

temp_columns = self.index

local_transpose = self._map_partitions(

lambda df: df.transpose(*args, **kwargs), index=temp_index)

local_transpose.columns = temp_columns

# Sum will collapse the NAs from the groupby

df = local_transpose.reduce_by_index(

lambda df: df.apply(lambda x: x), axis=1)

# Reassign the columns within partition to self.index.

# We have to use _depoly_func instead of _map_partition due to

# new_labels argument

def _reassign_columns(df, new_labels):

df.columns = new_labels

return df

df._df = [

_deploy_func.remote(

_reassign_columns,

part,

self.index) for part in df._df]

return df

T = property(transpose)

def dropna(self, axis, how, thresh=None, subset=[], inplace=False):

"""Create a new DataFrame from the removed NA values from this one.

Args:

axis (int, tuple, or list): The axis to apply the drop.

how (str): How to drop the NA values.

'all': drop the label if all values are NA.

'any': drop the label if any values are NA.

thresh (int): The minimum number of NAs to require.

subset ([label]): Labels to consider from other axis.

inplace (bool): Change this DataFrame or return a new DataFrame.

True: Modify the data for this DataFrame, return None.

False: Create a new DataFrame and return it.

Returns:

If inplace is set to True, returns None, otherwise returns a new

DataFrame with the dropna applied.

"""

raise NotImplementedError("Not yet")

def add(self, other, axis='columns', level=None, fill_value=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def agg(self, func, axis=0, *args, **kwargs):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def aggregate(self, func, axis=0, *args, **kwargs):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def align(self, other, join='outer', axis=None, level=None, copy=True,

fill_value=None, method=None, limit=None, fill_axis=0,

broadcast_axis=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def all(self, axis=None, bool_only=None, skipna=None, level=None,

**kwargs):

"""Return whether all elements are True over requested axis

Note:

If axis=None or axis=0, this call applies df.all(axis=1)

to the transpose of df.

"""

if axis is None or axis == 0:

df = self.T

axis = 1

else:

df = self

mapped = df._map_partitions(lambda df: df.all(axis,

bool_only,

skipna,

level,

**kwargs))

return to_pandas(mapped)

def any(self, axis=None, bool_only=None, skipna=None, level=None,

**kwargs):

"""Return whether all elements are True over requested axis

Note:

If axis=None or axis=0, this call applies df.all(axis=1)

to the transpose of df.

"""

if axis is None or axis == 0:

df = self.T

axis = 1

else:

df = self

mapped = df._map_partitions(lambda df: df.any(axis,

bool_only,

skipna,

level,

**kwargs))

return to_pandas(mapped)

def append(self, other, ignore_index=False, verify_integrity=False):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def apply(self, func, axis=0, broadcast=False, raw=False, reduce=None,

args=(), **kwds):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def as_blocks(self, copy=True):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def as_matrix(self, columns=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def asfreq(self, freq, method=None, how=None, normalize=False,

fill_value=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def asof(self, where, subset=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def assign(self, **kwargs):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def astype(self, dtype, copy=True, errors='raise', **kwargs):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def at_time(self, time, asof=False):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def between_time(self, start_time, end_time, include_start=True,

include_end=True):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def bfill(self, axis=None, inplace=False, limit=None, downcast=None):

"""Synonym for DataFrame.fillna(method='bfill')

"""

new_df = self.fillna(

method='bfill', axis=axis, limit=limit, downcast=downcast

)

if inplace:

self._df = new_df._df

self.columns = new_df.columns

else:

return new_df

def bool(self):

"""Return the bool of a single element PandasObject.

This must be a boolean scalar value, either True or False. Raise a

ValueError if the PandasObject does not have exactly 1 element, or that

element is not boolean

"""

shape = self.shape

if shape != (1,) and shape != (1, 1):

raise ValueError("""The PandasObject does not have exactly

1 element. Return the bool of a single

element PandasObject. The truth value is

ambiguous. Use a.empty, a.item(), a.any()

or a.all().""")

else:

return to_pandas(self).bool()

def boxplot(self, column=None, by=None, ax=None, fontsize=None, rot=0,

grid=True, figsize=None, layout=None, return_type=None,

**kwds):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def clip(self, lower=None, upper=None, axis=None, inplace=False, *args,

**kwargs):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def clip_lower(self, threshold, axis=None, inplace=False):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def clip_upper(self, threshold, axis=None, inplace=False):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def combine(self, other, func, fill_value=None, overwrite=True):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def combine_first(self, other):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def compound(self, axis=None, skipna=None, level=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def consolidate(self, inplace=False):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def convert_objects(self, convert_dates=True, convert_numeric=False,

convert_timedeltas=True, copy=True):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def corr(self, method='pearson', min_periods=1):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def corrwith(self, other, axis=0, drop=False):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def count(self, axis=0, level=None, numeric_only=False):

if axis == 1:

return self.T.count(axis=0,

level=level,

numeric_only=numeric_only)

else:

temp_index = [idx

for _ in range(len(self._df))

for idx in self.columns]

collapsed_df = sum(

ray.get(

self._map_partitions(

lambda df: df.count(

axis=axis,

level=level,

numeric_only=numeric_only),

index=temp_index)._df))

return collapsed_df

def cov(self, min_periods=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def cummax(self, axis=None, skipna=True, *args, **kwargs):

"""Perform a cumulative maximum across the DataFrame.

Args:

axis (int): The axis to take maximum on.

skipna (bool): True to skip NA values, false otherwise.

Returns:

The cumulative maximum of the DataFrame.

"""

if axis == 1:

return self._map_partitions(

lambda df: df.cummax(axis=axis, skipna=skipna,

*args, **kwargs))

else:

local_max = [_deploy_func.remote(

lambda df: pd.DataFrame(df.max()).T, self._df[i])

for i in range(len(self._df))]

new_df = DataFrame(local_max, self.columns)

last_row_df = pd.DataFrame([df.iloc[-1, :]

for df in ray.get(new_df._df)])

cum_df = [_prepend_partitions.remote(last_row_df, i, self._df[i],

lambda df:

df.cummax(axis=axis,

skipna=skipna,

*args, **kwargs))

for i in range(len(self._df))]

final_df = DataFrame(cum_df, self.columns)

return final_df

def cummin(self, axis=None, skipna=True, *args, **kwargs):

"""Perform a cumulative minimum across the DataFrame.

Args:

axis (int): The axis to cummin on.

skipna (bool): True to skip NA values, false otherwise.

Returns:

The cumulative minimum of the DataFrame.

"""

if axis == 1:

return self._map_partitions(

lambda df: df.cummin(axis=axis, skipna=skipna,

*args, **kwargs))

else:

local_min = [_deploy_func.remote(

lambda df: pd.DataFrame(df.min()).T, self._df[i])

for i in range(len(self._df))]

new_df = DataFrame(local_min, self.columns)

last_row_df = pd.DataFrame([df.iloc[-1, :]

for df in ray.get(new_df._df)])

cum_df = [_prepend_partitions.remote(last_row_df, i, self._df[i],

lambda df:

df.cummin(axis=axis,

skipna=skipna,

*args, **kwargs))

for i in range(len(self._df))]

final_df = DataFrame(cum_df, self.columns)

return final_df

def cumprod(self, axis=None, skipna=True, *args, **kwargs):

"""Perform a cumulative product across the DataFrame.

Args:

axis (int): The axis to take product on.

skipna (bool): True to skip NA values, false otherwise.

Returns:

The cumulative product of the DataFrame.

"""

if axis == 1:

return self._map_partitions(

lambda df: df.cumprod(axis=axis, skipna=skipna,

*args, **kwargs))

else:

local_prod = [_deploy_func.remote(

lambda df: pd.DataFrame(df.prod()).T, self._df[i])

for i in range(len(self._df))]

new_df = DataFrame(local_prod, self.columns)

last_row_df = pd.DataFrame([df.iloc[-1, :]

for df in ray.get(new_df._df)])

cum_df = [_prepend_partitions.remote(last_row_df, i, self._df[i],

lambda df:

df.cumprod(axis=axis,

skipna=skipna,

*args, **kwargs))

for i in range(len(self._df))]

final_df = DataFrame(cum_df, self.columns)

return final_df

def cumsum(self, axis=None, skipna=True, *args, **kwargs):

"""Perform a cumulative sum across the DataFrame.

Args:

axis (int): The axis to take sum on.

skipna (bool): True to skip NA values, false otherwise.

Returns:

The cumulative sum of the DataFrame.

"""

if axis == 1:

return self._map_partitions(

lambda df: df.cumsum(axis=axis, skipna=skipna,

*args, **kwargs))

else:

# first take the sum of each partition,

# append the sums of all previous partitions to current partition

# take cumsum and remove the appended rows

local_sum = [_deploy_func.remote(

lambda df: pd.DataFrame(df.sum()).T, self._df[i])

for i in range(len(self._df))]

new_df = DataFrame(local_sum, self.columns)

last_row_df = pd.DataFrame([df.iloc[-1, :]

for df in ray.get(new_df._df)])

cum_df = [_prepend_partitions.remote(last_row_df, i, self._df[i],

lambda df:

df.cumsum(axis=axis,

skipna=skipna,

*args, **kwargs))

for i in range(len(self._df))]

final_df = DataFrame(cum_df, self.columns)

return final_df

def describe(self, percentiles=None, include=None, exclude=None):

"""

Generates descriptive statistics that summarize the central tendency,

dispersion and shape of a dataset’s distribution, excluding NaN values.

Args:

percentiles (list-like of numbers, optional):

The percentiles to include in the output.

include: White-list of data types to include in results

exclude: Black-list of data types to exclude in results

Returns: Series/DataFrame of summary statistics

"""

obj_columns = [self.columns[i]

for i, t in enumerate(self.dtypes)

if t == np.dtype('O')]

rdf = self.drop(columns=obj_columns)

transposed = rdf.T

count_df = rdf.count()

mean_df = transposed.mean(axis=1)

std_df = transposed.std(axis=1)

min_df = to_pandas(rdf.min())

if percentiles is None:

percentiles = [.25, .50, .75]

percentiles_dfs = [transposed.quantile(q, axis=1)

for q in percentiles]

max_df = to_pandas(rdf.max())

describe_df = pd.DataFrame()

describe_df['count'] = count_df

describe_df['mean'] = mean_df

describe_df['std'] = std_df

describe_df['min'] = min_df

for i in range(len(percentiles)):

percentile_str = "{0:.0f}%".format(percentiles[i]*100)

describe_df[percentile_str] = percentiles_dfs[i]

describe_df['max'] = max_df

return describe_df.T

def diff(self, periods=1, axis=0):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def div(self, other, axis='columns', level=None, fill_value=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def divide(self, other, axis='columns', level=None, fill_value=None):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def dot(self, other):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def drop(self, labels=None, axis=0, index=None, columns=None, level=None,

inplace=False, errors='raise'):

"""Return new object with labels in requested axis removed.

Args:

labels: Index or column labels to drop.

axis: Whether to drop labels from the index (0 / 'index') or

columns (1 / 'columns').

index, columns: Alternative to specifying axis (labels, axis=1 is

equivalent to columns=labels).

level: For MultiIndex

inplace: If True, do operation inplace and return None.

errors: If 'ignore', suppress error and existing labels are

dropped.

Returns:

dropped : type of caller

"""

# inplace = validate_bool_kwarg(inplace, "inplace")

if labels is not None:

if index is not None or columns is not None:

raise ValueError("Cannot specify both 'labels' and "

"'index'/'columns'")

elif index is None and columns is None:

raise ValueError("Need to specify at least one of 'labels', "

"'index' or 'columns'")

new_df = self

is_axis_zero = axis is None or axis == 0 or axis == 'index'\

or axis == 'rows'

try:

if (is_axis_zero and columns is None) or index is not None:

values = labels if labels is not None else index

try:

try:

if len(values) == 0:

if inplace:

return

else:

return self

filtered_index = self._index.loc[list(values)]

except TypeError:

filtered_index = self._index.loc[[values]]

except KeyError:

raise ValueError(

"{} is not contained in the index".format(labels))

filtered_index.dropna(inplace=True)

partition_idx = [

filtered_index.loc[

filtered_index['partition'] == i

]['index_within_partition']

for i in range(len(self._df))

]

new_df = [

_deploy_func.remote(

lambda df, new_labels: df.drop(

new_labels, level=level, errors='ignore'),

self._df[i], partition_idx[i]

)

for i in range(len(self._df))

]

new_index = self._index.copy().drop(values, errors=errors)

new_df = DataFrame(new_df, self.columns, index=new_index.index)

except (ValueError, KeyError):

if errors == 'raise':

raise

new_df = self

try:

if not is_axis_zero or columns is not None:

values = labels if labels else columns

new_df = new_df._map_partitions(

lambda df: df.drop(

values, axis=1, level=level, errors='ignore')

)

new_columns = self.columns.to_series().drop(values,

errors=errors)

new_df.columns = pd.Index(new_columns)

except (ValueError, KeyError):

if errors == 'raise':

raise

new_df = self

if inplace:

self._update_inplace(

df=new_df._df,

index=new_df.index,

columns=new_df.columns

)

else:

return new_df

def drop_duplicates(self, subset=None, keep='first', inplace=False):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")

def duplicated(self, subset=None, keep='first'):

raise NotImplementedError(

"To contribute to Pandas on Ray, please visit "

"github.com/ray-project/ray.")