Repository: dask/dask
Pull request: Add Dataframe.replace
Fixes #4573

- [ ] Tests added / passed
- [ ] Passes `flake8 dask`


Problem statement:
pandas .replace() method in dask
Hi All,

I would like to recode a column variable using dask. It could be done in pandas using .replace(). An example code is below.

in pandas:
```python
months = {'jan':1,'feb':2,'mar':3,'apr':4,'may':5,'jun':6,'jul':7,'aug':8,'sep':9,'oct':10,'nov':11,'dec':12}
df.replace(months)
```
Would anyone able to tell me whether an equivalent method in dask. 

Thanks

Michael

Hints from issue discussion:
Does `DataFrame.map_partitions(pandas.DataFrame.replace, months)` work?

This should be relatively easy to add to dask dataframe.

On Sat, Mar 9, 2019 at 2:32 AM MichaelSchroter <notifications@github.com>
wrote:

> Hi All,
>
> I would like to recode a column variable using dask. It could be done in
> pandas using .replace(). An example code is below.
>
> in pandas:
>
> months = {'jan':1,'feb':2,'mar':3,'apr':4,'may':5,'jun':6,'jul':7,'aug':8,'sep':9,'oct':10,'nov':11,'dec':12}
> df.replace(months)
>
> Would anyone able to tell me whether an equivalent method in dask.
>
> Thanks
>
> Michael
>
> —
> You are receiving this because you are subscribed to this thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/issues/4573>, or mute the thread
> <https://github.com/notifications/unsubscribe-auth/ABQHIp64bejfWi692ctRgqA3RdlSLGYEks5vU3GXgaJpZM4bmjuP>
> .
>

Hi Tom,

Here is my code below.
```python
from dask.distributed import Client
client = Client()
import dask.dataframe as dd
import numpy as np

in[]:
bank = dd.read_parquet('/home/michael/bank/bank.parquet') #Simple small datafile
bank['month'].value_counts().compute() #show how data is represented

out[]: 
may    2824
aug    1519
jul    1514
jun    1222
nov     943
apr     923
feb     776
oct     392
jan     344
sep     319
mar     276
dec     110
Name: month, dtype: int64

in[]:
months = {'jan':1,'feb':2,'mar':3,'apr':4,'may':5,'jun':6,'jul':7,'aug':8,'sep':9,'oct':10,'nov':11,'dec':12}
bank.map_partitions(bank.replace, months) #intends to search where ever there is a 'jan' and changes it into 1.
out[]:
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-7-a16e574df098> in <module>
----> 1 bank.map_partitions(bank.replace, months)

~/env1/lib/python3.5/site-packages/dask/dataframe/core.py in __getattr__(self, key)
   2558             return self[key]
   2559         else:
-> 2560             raise AttributeError("'DataFrame' object has no attribute %r" % key)
   2561 
   2562     def __dir__(self):

AttributeError: 'DataFrame' object has no attribute 'replace'

in[]:
bank['month'].map_partitions(bank['month'].replace, months) #Searches for the name 'jan' in column 'month' and changes it into 1.

out[]:
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-5-7084694c8d5c> in <module>
----> 1 bank['month'].map_partitions(bank['month'].replace, months)

AttributeError: 'Series' object has no attribute 'replace'
```

What worked:
```python
bank['month'] = bank['month'].mask(bank['month'] == 'jan', 1).mask(bank['month'] == 'feb', 2).mask(bank['month'] == 'mar', 3).mask(bank['month'] == 'apr', 4).mask(bank['month'] == 'may', 5).mask(bank['month'] == 'jun', 6).mask(bank['month'] == 'jul', 7).mask(bank['month'] == 'aug', 8).mask(bank['month'] == 'sep', 9).mask(bank['month'] == 'oct', 10).mask(bank['month'] == 'nov', 11).mask(bank['month'] == 'dec', 12) #maybe little cumbersome but worked. Not sure whether there is a better solution for this. 
```

Thanks

Michael
 


Tom's suggestion was to use the `pandas.DataFrame.replace` method, not `bank.replace`.  Copying his suggestion from above

    df.map_partitions(pandas.DataFrame.replace, months)

Also, for better help you may want to make an mcve.  You may want to read through http://matthewrocklin.com/blog/work/2018/02/28/minimal-bug-reports .
Hi @mrocklin,

I have a dask dataframe. Not a pandas one. (the code mentions a pandas dataframe )Would you be able to tell me where should I include my dask dataframe in the code that Tom mentioned. For example I have a dask dataframe named `ddf` where should I include it in the code that @TomAugspurger  mentioned.

Thanks

Michael
your dataframe is the `df` object
Any update here @MichaelSchroter? Did @TomAugspurger suggestion of using `ddf.map_partitions(pandas.DataFrame.replace, months)` address your issue? 
> Any update here @MichaelSchroter? Did @TomAugspurger suggestion of using `ddf.map_partitions(pandas.DataFrame.replace, months)` address your issue?

Hi @jrbourbeau,

What I have are `dask` dataframes. Within the brackets there is a `pandas` dataframe. The reason I did not want to convert into pandas is that the the dataframe is large. Furthermore, the advantage of using dask is lost if the dataframe could load in memory in my instance. The workaround that I used was.

```python
for to_replace in list_:
    ddf = ddf.mask(ddf == to_replace, 'change_to_this_value') 
```
This worked for me. The for loop resolves the issue when the `to_replace` value is a list. i.e. `.replace(to_replace, value)`
I do not know any other solution in dask for pandas `.replace()` method.

Thanks

Michael
> This worked for me

Glad to hear you found something that addressed your problem! 

> I did not want to convert into pandas is that the the dataframe is large

That's totally understandable. However, calling `map_partitions` with a function (in this case the function is `pd.DataFrame.replace`) on a Dask DataFrame does not load the Dask DataFrame into memory. Hopefully, the below example helps illustrates that point. 

```python
import pandas as pd
import dask.dataframe as dd

# Construct pandas DataFrame
df = pd.DataFrame({'A': [0, 1, 2, 3, 4],
                   'B': [5, 6, 7, 8, 9],
                   'C': ['a', 'b', 'c', 'd', 'e']})

# Convert to Dask DataFrame
ddf = dd.from_pandas(df, npartitions=3)

# Do replacement on Dask DataFrame
to_replace = {0: 10, 1: 100}
result = ddf.map_partitions(pd.DataFrame.replace, to_replace)
# Note that result is a Dask DataFrame
print(result)
```

Running the above example outputs:

```
Dask DataFrame Structure:
                   A      B       C
npartitions=2
0              int64  int64  object
2                ...    ...     ...
4                ...    ...     ...
Dask Name: replace, 4 tasks
```
I'm going to close this issue as it appears both the `mask` and `map_partitions` approaches address the originally posted issue. Feel free to re-open if this is not the case. 
Hi @jrbourbeau,

Ok thanks a lot. That also helped. Your solution is much more efficient.

Thanks

Michael

Commit message: Add Dataframe.replace

Fixes #4573

Patch to review:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 86a5a300ae1..d031528d253 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1109,6 +1109,10 @@ def sample(self, n=None, frac=None, replace=False, random_state=None):
         graph = HighLevelGraph.from_collections(name, dsk, dependencies=[self])
         return new_dd_object(graph, name, self._meta, self.divisions)
 
+    def replace(self, to_replace=None, value=None, regex=False):
+        return self.map_partitions(M.replace, to_replace=to_replace,
+                                   value=value, regex=regex)
+
     def to_dask_array(self, lengths=None):
         """Convert a dask DataFrame to a dask array.
 
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 85238922be8..e6fd091f692 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -3527,3 +3527,13 @@ def test_index_divisions():
     assert_eq(ddf.index + 1, df.index + 1)
     assert_eq(10 * ddf.index, 10 * df.index)
     assert_eq(-ddf.index, -df.index)
+
+
+def test_replace():
+    df = pd.DataFrame({'x': [1, 2, 3, 4, 5]})
+    ddf = dd.from_pandas(df, npartitions=2)
+
+    assert_eq(df.replace(1, 10), ddf.replace(1, 10))
+    assert_eq(df.replace({1: 10, 2: 20}), ddf.replace({1: 10, 2: 20}))
+    assert_eq(df.x.replace(1, 10), ddf.x.replace(1, 10))
+    assert_eq(df.x.replace({1: 10, 2: 20}), ddf.x.replace({1: 10, 2: 20}))


Code review comments:
diff_hunk: @@ -1109,6 +1109,10 @@ def sample(self, n=None, frac=None, replace=False, random_state=None):
         graph = HighLevelGraph.from_collections(name, dsk, dependencies=[self])
         return new_dd_object(graph, name, self._meta, self.divisions)
 
+    def replace(self, to_replace=None, value=None, regex=False):
path: dask/dataframe/core.py
text: @user2:
```suggestion
    @user1(pd.DataFrame)
    def replace(self, to_replace=None, value=None, regex=False):
```