add to_castra method to DataFrame

This gives us a nice efficient store for DataFrames

dask/dataframe/core.py

@@ -629,6 +629,22 @@ def assign(self, **kwargs):
 
         return elemwise(_assign, self, *pairs, columns=list(df2.columns))
 
+    def to_castra(self, fn=None, categories=None):
+        """ Write DataFrame to Castra on-disk store
+
+        See https://github.com/blosc/castra for details
+
+        See Also:
+            Castra.to_dask
+        """
+        from castra import Castra
+        name = 'to-castra' + next(tokens)
+        dsk = {name: (Castra, fn, (self._name, 0), categories)}
+        dsk.update(dict(((name, i), (Castra.extend, name, (self._name, i)))
+                        for i in range(self.npartitions)))
+        c, _ = get(merge(dsk, self.dask), [name, list(dsk.keys())])
+        return c
+
 
 def _assign(df, *pairs):
     kwargs = dict(partition(2, pairs))

dask/dataframe/tests/test_io.py

@@ -394,3 +394,13 @@ def test_from_dask_array_raises():
     except Exception as e:
         assert 'hello' in str(e)
         assert '3' in str(e)
+
+
+def test_to_castra():
+    pytest.importorskip('castra')
+    df = pd.DataFrame({'x': ['a', 'b', 'c', 'D'],
+                       'y': [1, 2, 3, 4]})
+    a = dd.from_pandas(df, 2)
+
+    c = a.to_castra()
+    assert eq(a, c[:])