Dask distributed cluster, 'str' object has no attribute 'apply' error during computations

Original question on Stackoverflow - https://stackoverflow.com/questions/52994892/dask-distributed-cluster-str-object-has-no-attribute-apply-error-during-com . I decided to post it here because, this behaviour looks like a bug.

I have a dynamic Dask Kubernetes cluster on GCloud based on this repo
https://github.com/VMois/dask-k8s-chart.

During data processing (20 parquet files, each file about 80MB stored on Gcloud storage) using `.apply()`
I sometimes receive an error from the cluster:

File "/usr/local/lib/python3.5/site-packages/dask/dataframe/core.py", line 3660, in apply_and_enforce
File "/usr/local/lib/python3.5/site-packages/dask/utils.py", line 688, in __call__
AttributeError: 'str' object has no attribute 'apply'

My func to generate test data:

```python
file_bytes = open('test_img.jpg', 'rb').read() # about 59 KB
files_count = 20
file_ids = [{'id': 10 + file_id, 'data': file_bytes} for file_id in range(files_count)]

def cluster_data_gen(args):
amount = 1500
face_data = args['data']
file_id = args['id']
files_ids = [int(file_id) for x in range(amount)]
file_bytes = [bytes(face_data) for x in range(amount)]
df = pd.DataFrame({
'file_id': videos_ids,
'file': files_bytes,
})
df = dd.from_pandas(df, npartitions=1)
df.to_parquet('gcs://test/files/video_{}.parquet'.format(file_id),
storage_options={'token': 'cloud'},
object_encoding={
'file': 'bytes',
'file_id': 'int'
},
compute=True)

x = client.map(cluster_data_gen, file_ids)
client.gather(x)
```

My func to process test data (where error occurs):
```python
client = Client('')
client.restart()
df = dask.delayed(dd.read_parquet)('gcs://test/files/video_{}.parquet/*.parquet',
storage_options={'token': 'cloud'},
engine='fastparquet')
df = dask.compute(df)[0]

def test(row):
img_bytes = row['img']
img = cv2.imdecode(np.frombuffer(img_bytes, np.uint8), -1)
del img_bytes
img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
img = cv2.imencode('.jpg', img)[1]
return pd.Series({'img': img.tobytes(), 'video_id': row['video_id']})

df = df.apply(test, axis=1, meta={'img': 'bytes', 'video_id': 'int'})
df.to_parquet('gcs://test/result/',
storage_options={'token': 'cloud'},
object_encoding={'img': 'bytes', 'video_id': 'int'},
compute=True)
```

Some additional info:

- each worker is Kubernetes pod with 0.5 CPU and 1.5GB RAM.
- cluster can scale up to 5 workers.

I don't think the issue is in data, because sometimes an error occurs, sometimes not.

I have tried to remove heavy computations from `apply()` and return dummy data, like this:

def test(row):
img = b'a'
return pd.Series({'img': img, 'video_id': 1})

to reduce a load on the cluster, but it doesn't change anything. The cluster still sometimes returns the same error.

Do you have any idea why this error occurs?
Maybe, you can suggest further directions to explore why it happens? Thank you very much!

Dask version: 0.19.1

Dask distributed version: 1.23.1

Python version: 3.5.1