Rewrite of dask interface.

* Consider data locality in distributing data.
* New interface that looks similar to the one with single node.

demo/dask/demo.py

@@ -0,0 +1,24 @@
+import xgboost as xgb
+from xgboost.distributed import DaskDMatrix
+from dask.distributed import Client
+from dask.distributed import LocalCluster
+from dask import dataframe as dd
+
+
+if __name__ == '__main__':
+    cluster = LocalCluster(n_workers=4)  # or use any other clusters
+    client = Client(cluster)
+
+    train_csv = dd.read_csv('./train.csv')
+    ad_feature = dd.read_csv('./adFeature.csv')
+    train_csv = train_csv.merge(ad_feature, how='left', on=['aid'])
+
+    label = train_csv['label']
+    train_csv = train_csv.drop(labels=['label'], axis=1)
+
+    dtrain = DaskDMatrix(train_csv, label)
+
+    bst = xgb.train({'verbosity': 3}, dtrain)
+
+    with open('result', 'w') as fd:
+        fd.write(str(bst))

python-package/xgboost/__init__.py

@@ -11,7 +11,6 @@
 from .core import DMatrix, Booster
 from .training import train, cv
 from . import rabit                   # noqa
-from . import dask  # noqa
 from . import tracker  # noqa
 from .tracker import RabitTracker  # noqa
 try:

python-package/xgboost/core.py

@@ -1040,7 +1040,7 @@ def __init__(self, params=None, cache=(), model_file=None):
         """
         for d in cache:
             if not isinstance(d, DMatrix):
-                raise TypeError('invalid cache item: {}'.format(type(d).__name__))
+                raise TypeError('invalid cache item: {}'.format(type(d).__name__), cache)
             self._validate_features(d)
 
         dmats = c_array(ctypes.c_void_p, [d.handle for d in cache])