GridSearch Error: KeyError: 'data'

I am getting the following error when running a gridsearch on dask distributed back-end. This error is nonexistent when just running sklearn gridsearch on single core local machine. I don't know where that KeyError is coming from; I don't have anything in my pipeline that references the key 'data'.

Here is the full error traceback I am getting:

```
[Parallel(n_jobs=-1)]: Using backend DaskDistributedBackend with 38 concurrent workers.
distributed.client - ERROR - Error in callback .callback_wrapper at 0x11c4a8f28> of :
Traceback (most recent call last):
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/client.py", line 285, in execute_callback
fn(fut)
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/joblib/_dask.py", line 260, in callback_wrapper
result = future.result()
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/client.py", line 217, in result
result = self.client.sync(self._result, callback_timeout=timeout, raiseit=False)
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/client.py", line 780, in sync
self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/utils.py", line 348, in sync
raise exc.with_traceback(tb)
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/utils.py", line 332, in f
result[0] = yield future
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/tornado/gen.py", line 735, in run
value = future.result()
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/client.py", line 242, in _result
result = await self.client._gather([self])
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/client.py", line 1781, in _gather
response = await future
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/client.py", line 1832, in _gather_remote
response = await retry_operation(self.scheduler.gather, keys=keys)
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/utils_comm.py", line 391, in retry_operation
operation=operation,
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/utils_comm.py", line 379, in retry
return await coro()
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/core.py", line 757, in send_recv_from_rpc
result = await send_recv(comm=comm, op=key, **kwargs)
File "/Users/rihun/anaconda3/envs/dask_env/lib/python3.7/site-packages/distributed/core.py", line 556, in send_recv
raise exc.with_traceback(tb)
File "/opt/conda/lib/python3.7/site-packages/distributed/core.py", line 412, in handle_comm
File "/opt/conda/lib/python3.7/site-packages/distributed/scheduler.py", line 2792, in gather
File "/opt/conda/lib/python3.7/site-packages/distributed/utils_comm.py", line 87, in gather_from_workers
KeyError: 'data'
```
Sample Dataset

entity_name
classification


great tech
other


xfone communication ltd
other


coventrys
other


pt invensys indonesia
other


massillon cable tv inc
other


city of New York
government


police department
government


ministry of finance
government


US Navy
military


US Army
military


AFB
military

Code Example

```
import sys
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.pipeline import FeatureUnion
from sklearn.linear_model import LogisticRegression
import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.svm import SVC
import re
from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
import time
from sklearn.model_selection import train_test_split
from sklearn.utils import shuffle
import string
from dask.distributed import Client
import joblib
import logging
from sklearn.model_selection import StratifiedKFold

data = pd.read_csv('https://raw.githubusercontent.com/rileyhun/dask/master/training_data_03_23_u.csv')
X_train, X_test, y_train, y_test = train_test_split(data['entity_name'], data['classification'], test_size=0.3, random_state=123)

vec_transformer = FeatureUnion([
('word_name',
Pipeline([
('tfidf',
TfidfVectorizer(sublinear_tf=False,
smooth_idf=False,
use_idf=1,
min_df=2,
preprocessor=lambda x: re.sub("[" + string.punctuation + " +" + "]+", " ", x.lower()),
analyzer='char_wb',
token_pattern=r'\S+',
ngram_range=(2,10),
dtype=np.float32))]))
])

pipeline = Pipeline([
('vectorizer', vec_transformer),
('model', LogisticRegression())
])

client = Client(':')

param_grid = {
"model__C": [1, 3],
"model__tol": [0.001, 0.01]
}

clf = GridSearchCV(pipeline,
param_grid,
verbose=8,
cv=3,
scoring='f1_weighted',
refit=True)

with joblib.parallel_backend('dask'):
clf.fit(X_train, y_train)
```

There are no conflicts between scheduler, client and the dask workers.