Repository: dask/dask

Issue:
Implement Series/DataFrame.attrs
Like xarray, pandas supports attaching arbitrary metadata to DataFrames and persisting it across operations. https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.attrs.html

Dask could pretty easily implement this as well. We'd have a 

* `_Frame.attrs` property. This would likely return `self._meta.attrs`.
* We'd verify that `dd.from_pandas(data)` correctly extracts `attrs` from `data` if it's a pandas NDFrame (I don't see docs for attrs in cudf)
* A few sanity tests making sure that it's propagated in certain operations.

Most of the hard work of propagating attrs will be handled by executing the pandas functions on `_meta`, which should propagate it. There may be some issues around user-provided `meta`. But those can be handled separately.


Discussion:
I gave this a try. One thing that I found rather annoying though is that the Series does not keep their attrs.

```python
import pandas as pd 
import dask.dataframe as dd

df = pd.DataFrame({"A": [1, 2], "B": [3, 4], "C": [5, 6]})
df.attrs = {'date': '2020-10-16'}
df.A.attrs['unit'] = 1
ddf = dd.from_pandas(df, 2)

ddf.attrs
Out[36]: {'date': '2020-10-16'}
ddf.A.attrs
Out[37]: {}
```
This is because at initialization of `_Frame`, meta is cleaned from any rows of data using`make_meta_pandas(x, index=None)`, which uses `x.iloc[:0] `to filter out the data. 

I suppose this is an upstream error because pandas does not keep the attrs when using that method either:
```python
df.A.attrs
Out[38]: {'unit': 1}

df.iloc[:0].A.attrs
Out[39]: {}
```

Are there any other ways of filtering away the data in pandas that keeps the attrs?
@Illviljan you can report lost attrs at https://github.com/pandas-dev/pandas/issues/28283 if it's not included in the list already. I don't think we'll want to implement any workarounds in Dask.

Patch:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -331,6 +331,15 @@ def npartitions(self):
         """Return number of partitions"""
         return len(self.divisions) - 1
 
+    @property
+    @derived_from(pd.DataFrame)
+    def attrs(self):
+        return self._meta.attrs
+
+    @attrs.setter
+    def attrs(self, value):
+        self._meta.attrs = dict(value)
+
     @property
     def size(self):
         """Size of the Series or DataFrame as a Delayed object.
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -398,6 +398,8 @@ def meta_nonempty_dataframe(x):
         data[i] = dt_s_dict[dt]
     res = pd.DataFrame(data, index=idx, columns=np.arange(len(x.columns)))
     res.columns = x.columns
+    if PANDAS_GT_100:
+        res.attrs = x.attrs
     return res
 
 
@@ -586,7 +588,10 @@ def _nonempty_series(s, idx=None):
         entry = _scalar_from_dtype(dtype)
         data = np.array([entry, entry], dtype=dtype)
 
-    return pd.Series(data, name=s.name, index=idx)
+    out = pd.Series(data, name=s.name, index=idx)
+    if PANDAS_GT_100:
+        out.attrs = s.attrs
+    return out
 
 
 def is_dataframe_like(df):


Test patch:
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -4429,3 +4429,37 @@ def test_fuse_roots():
     res = ddf1.where(ddf2)
     hlg = fuse_roots(res.__dask_graph__(), keys=res.__dask_keys__())
     hlg.validate()
+
+
+@pytest.mark.skipif(not dd._compat.PANDAS_GT_100, reason="attrs introduced in 1.0.0")
+def test_attrs_dataframe():
+    df = pd.DataFrame({"A": [1, 2], "B": [3, 4], "C": [5, 6]})
+    df.attrs = {"date": "2020-10-16"}
+    ddf = dd.from_pandas(df, 2)
+
+    assert df.attrs == ddf.attrs
+    assert df.abs().attrs == ddf.abs().attrs
+
+
+@pytest.mark.skipif(not dd._compat.PANDAS_GT_100, reason="attrs introduced in 1.0.0")
+def test_attrs_series():
+    s = pd.Series([1, 2], name="A")
+    s.attrs["unit"] = "kg"
+    ds = dd.from_pandas(s, 2)
+
+    assert s.attrs == ds.attrs
+    assert s.fillna(1).attrs == ds.fillna(1).attrs
+
+
+@pytest.mark.skipif(not dd._compat.PANDAS_GT_100, reason="attrs introduced in 1.0.0")
+@pytest.mark.xfail(reason="df.iloc[:0] does not keep the series attrs")
+def test_attrs_series_in_dataframes():
+    df = pd.DataFrame({"A": [1, 2], "B": [3, 4], "C": [5, 6]})
+    df.A.attrs["unit"] = "kg"
+    ddf = dd.from_pandas(df, 2)
+
+    # Fails because the pandas iloc method doesn't currently persist
+    # the attrs dict for series in a dataframee. Dask uses df.iloc[:0]
+    # when creating the _meta dataframe in make_meta_pandas(x, index=None).
+    # Should start xpassing when df.iloc works. Remove the xfail then.
+    assert df.A.attrs == ddf.A.attrs