PCA+pipeline+GridSearchCV error

There seems to be an issue with sklearn PCA+pipeline and dask_ml gridsearchCV. Please see my example below. Apologies if I am totally missing something.

Relevant Versions:

```
dask: 2.12.0
dask_ml: 1.2.0
sklearn: 0.22.2.post1
```

### Minimal Example:

The following code shows that
* Pipeline+PCA+Sklearn_GridSearchCV works
* Pipeline+SelectKBest+Dask_GridSearchCV works
* Pipeline+PCA+Dask_GridSearchCV fails

```python
from dask.distributed import Client, LocalCluster
from dask_ml.model_selection import GridSearchCV as dask_GS
from sklearn.model_selection import GridSearchCV as sk_GS
from sklearn.datasets import make_multilabel_classification
from sklearn import svm
from sklearn.pipeline import Pipeline
from sklearn.decomposition import PCA
from sklearn.feature_selection import SelectKBest, f_classif
import numpy as np

cluster = LocalCluster(n_workers=2,threads_per_worker=2)
client = Client(cluster)

X, Y = make_multilabel_classification(n_classes=12, n_labels=1,n_features=271,
n_samples=1200,
random_state=1)
Y = Y.sum(axis=1)

N_FEATURES_OPTIONS_pca = np.arange(10)[1::3]
N_FEATURES_OPTIONS_sel = np.arange(10)[1::3]
Cs = [1,10,100.]
gammas = [.001,0.01]

pca = PCA(iterated_power='auto')
selection = SelectKBest(f_classif)
svc = svm.SVC()

pipe1 = Pipeline([
('reduce_dim', selection),
('classify', svc)])

pipe2 = Pipeline([
('reduce_dim', pca),
('classify', svc)])

param_grid1 = [{'reduce_dim__k': N_FEATURES_OPTIONS_sel,
'classify__C': Cs,
'classify__gamma': gammas}]
param_grid2 = [{'reduce_dim__n_components': N_FEATURES_OPTIONS_pca,
'classify__C': Cs,
'classify__gamma': gammas}]

#Sklearn Gridsearch with PCA pipeline
sk_clf = sk_GS(pipe2, param_grid2,cv=3,scoring='f1_macro',refit=True)
sk_clf.fit(X,Y)
print(sk_clf.best_score_)

#Dask Gridsearch with SelectKbest
dask_clf1 = dask_GS(pipe1, param_grid1,cv=3,scheduler=client,scoring='f1_macro',refit=True)
dask_clf1.fit(X,Y)
print(dask_clf1.best_score_)

#Dask Gridsearch with PCA
dask_clf2 = dask_GS(pipe2, param_grid2,cv=3,scheduler=client,scoring='f1_macro',refit=True)
dask_clf2.fit(X,Y)
print(dask_clf2.best_score_)

```

This results in several core dump files and the following error:

Results In:

```
0.18358307544807187
0.16009973534039232
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
distributed.nanny - WARNING - Restarting worker
('score-f44a4381cb4779b9d45ba2c0ba7c2a72', 15, 1) has failed... retrying
---------------------------------------------------------------------------
KeyError Traceback (most recent call last)
in
55 #Dask Gridsearch with PCA
56 dask_clf2 = dask_GS(pipe2, param_grid2,cv=3,scheduler=client,scoring='f1_macro',refit=True)
---> 57 dask_clf2.fit(X,Y)
58 print(dask_clf2.best_score_)

/opt/conda/envs/py_geo/lib/python3.7/site-packages/dask_ml/model_selection/_search.py in fit(self, X, y, groups, **fit_params)
1255 else:
1256 logger.warning("{} has failed... retrying".format(future.key))
-> 1257 future.retry()
1258 ac.add(future)
1259

/opt/conda/envs/py_geo/lib/python3.7/site-packages/distributed/client.py in retry(self, **kwargs)
307 Client.retry
308 """
--> 309 return self.client.retry([self], **kwargs)
310
311 def cancelled(self):

/opt/conda/envs/py_geo/lib/python3.7/site-packages/distributed/client.py in retry(self, futures, asynchronous)
2139 futures: list of Futures
2140 """
-> 2141 return self.sync(self._retry, futures, asynchronous=asynchronous)
2142
2143 @gen.coroutine

/opt/conda/envs/py_geo/lib/python3.7/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
778 else:
779 return sync(
--> 780 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
781 )
782

/opt/conda/envs/py_geo/lib/python3.7/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
346 if error[0]:
347 typ, exc, tb = error[0]
--> 348 raise exc.with_traceback(tb)
349 else:
350 return result[0]

/opt/conda/envs/py_geo/lib/python3.7/site-packages/distributed/utils.py in f()
330 if callback_timeout is not None:
331 future = asyncio.wait_for(future, callback_timeout)
--> 332 result[0] = yield future
333 except Exception as exc:
334 error[0] = sys.exc_info()

/opt/conda/envs/py_geo/lib/python3.7/site-packages/tornado/gen.py in run(self)
733
734 try:
--> 735 value = future.result()
736 except Exception:
737 exc_info = sys.exc_info()

/opt/conda/envs/py_geo/lib/python3.7/site-packages/distributed/client.py in _retry(self, futures)
2128 response = await self.scheduler.retry(keys=keys, client=self.id)
2129 for key in response:
-> 2130 st = self.futures[key]
2131 st.retry()
2132

KeyError: "('pca-fit-transform-f44a4381cb4779b9d45ba2c0ba7c2a72', 0, 1)"
```

Dask Distributed worker / scheduler logs

```
{'Scheduler': 'distributed.scheduler - INFO - Clear task state\n'
'distributed.scheduler - INFO - Scheduler at: '
'tcp://127.0.0.1:41727\n'
'distributed.scheduler - INFO - dashboard at: '
'127.0.0.1:8787\n'
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:35216\n'
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:44716\n'
'distributed.scheduler - INFO - Receive client connection: '
'Client-ee79b350-6fec-11ea-878c-0cc47a4279e3\n'
'distributed.scheduler - INFO - Remove worker \n"
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:38830\n'
'distributed.scheduler - INFO - Remove worker \n"
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:43574\n'
'distributed.scheduler - INFO - Remove worker \n"
'distributed.scheduler - INFO - Remove worker \n"
'distributed.scheduler - INFO - Lost all workers\n'
'distributed.scheduler - INFO - Client '
'Client-ee79b350-6fec-11ea-878c-0cc47a4279e3 requests to retry 1 '
'keys\n'
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:33853\n'
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:33057\n'
'distributed.scheduler - INFO - Remove worker \n"
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:33579\n'
'distributed.scheduler - INFO - Remove worker \n"
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:33304\n'
'distributed.scheduler - INFO - Remove worker \n"
'distributed.scheduler - INFO - Register worker \n"
'distributed.scheduler - INFO - Starting worker compute stream, '
'tcp://127.0.0.1:40951',
'tcp://127.0.0.1:33304': 'distributed.worker - INFO - Start worker '
'at: tcp://127.0.0.1:33304\n'
'distributed.worker - INFO - Listening '
'to: tcp://127.0.0.1:33304\n'
'distributed.worker - INFO - Waiting to connect '
'to: tcp://127.0.0.1:41727\n'
'distributed.worker - INFO - '
'-------------------------------------------------\n'
'distributed.worker - INFO - '
'Threads: 2\n'
'distributed.worker - INFO - '
'Memory: 6.50 GB\n'
'distributed.worker - INFO - Local Directory: '
'/project/cper_neon_aop/neon_temporal/dask-worker-space/worker-nswgxjil\n'
'distributed.worker - INFO - '
'-------------------------------------------------\n'
'distributed.worker - INFO - Registered '
'to: tcp://127.0.0.1:41727\n'
'distributed.worker - INFO - '
'-------------------------------------------------',
'tcp://127.0.0.1:40951': 'distributed.worker - INFO - Start worker '
'at: tcp://127.0.0.1:40951\n'
'distributed.worker - INFO - Listening '
'to: tcp://127.0.0.1:40951\n'
'distributed.worker - INFO - Waiting to connect '
'to: tcp://127.0.0.1:41727\n'
'distributed.worker - INFO - '
'-------------------------------------------------\n'
'distributed.worker - INFO - '
'Threads: 2\n'
'distributed.worker - INFO - '
'Memory: 6.50 GB\n'
'distributed.worker - INFO - Local Directory: '
'/project/cper_neon_aop/neon_temporal/dask-worker-space/worker-31u54x2o\n'
'distributed.worker - INFO - '
'-------------------------------------------------\n'
'distributed.worker - INFO - Registered '
'to: tcp://127.0.0.1:41727\n'
'distributed.worker - INFO - '
'-------------------------------------------------'}
```

Results of gdb on core dump file:

```
Core was generated by `/opt/conda/envs/py_geo/bin/python -c from multiprocessing.forkserver import mai'.
Program terminated with signal 11, Segmentation fault.
```

Start by running the minimal example with the reported Dask, dask-ml, and scikit-learn versions, then inspect dask_ml/model_selection/_search.py at the future.retry call and the distributed worker and scheduler logs. Done means PCA inside a pipeline completes under dask_ml GridSearchCV without worker restarts or the missing-future KeyError, while the SelectKBest and scikit-learn comparisons remain valid.

Written by the indexing model from the issue text.