Dask-Yarn & Dask-Dataframe map_partitions apply Error

Hi

I have a large python function that I wish to execute in parallel. The dataframe contains only the text data and an identifier. When I execute the following command to process the function in parallel it works fine.
```python
results = dd.from_pandas(dt,npartitions=nCores).map_partitions(lambda df1 : df1.apply(lambda x : proctxt(x['txt'], x['pm'],nlpe, nlpf))).compute(scheduler='processes')
```

When I try to run the same command using the dask-yarn interface, the command does not work:
```python
results = dd.from_pandas(dt,npartitions=nCores).map_partitions(lambda df1 : df1.apply(lambda x : proctxt(x['txt'], x['pm'],nlpe, nlpf))).compute(scheduler='distributed')
```
The error I get is the following:
```
Traceback (most recent call last):
File "pandas/_libs/index.pyx", line 162, in pandas._libs.index.IndexEngine.get_loc
File "pandas/_libs/hashtable_class_helper.pxi", line 958, in pandas._libs.hashtable.Int64HashTable.get_item
TypeError: an integer is required

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/usr/local/lib/python3.5/dist-packages/dask/dataframe/utils.py", line 137, in raise_on_meta_error
yield
File "/usr/local/lib/python3.5/dist-packages/dask/dataframe/core.py", line 3601, in _emulate
return func(*_extract_meta(args, True), **_extract_meta(kwargs, True))
File "fullscript.py", line 677, in
results = dd.from_pandas(dt,npartitions=nCores).map_partitions(lambda df1 : df1.apply(lambda x : proctxt(x['txt'], x['pm'], nlpe, nlpf))).compute(scheduler='distributed')
File "/usr/local/lib/python3.5/dist-packages/pandas/core/frame.py", line 6014, in apply
return op.get_result()
File "/usr/local/lib/python3.5/dist-packages/pandas/core/apply.py", line 318, in get_result
return super(FrameRowApply, self).get_result()
File "/usr/local/lib/python3.5/dist-packages/pandas/core/apply.py", line 142, in get_result
return self.apply_standard()
File "/usr/local/lib/python3.5/dist-packages/pandas/core/apply.py", line 248, in apply_standard
self.apply_series_generator()
File "/usr/local/lib/python3.5/dist-packages/pandas/core/apply.py", line 277, in apply_series_generator
results[i] = self.f(v)
File "fullscript.py", line 677, in
results = dd.from_pandas(dt,npartitions=nCores).map_partitions(lambda df1 : df1.apply(lambda x : proctxt(x['txt'], x['pm'], nlpe, nlpf))).compute(scheduler='distributed')
File "/usr/local/lib/python3.5/dist-packages/pandas/core/series.py", line 767, in __getitem__
result = self.index.get_value(self, key)
File "/usr/local/lib/python3.5/dist-packages/pandas/core/indexes/base.py", line 3118, in get_value
tz=getattr(series.dtype, 'tz', None))
File "pandas/_libs/index.pyx", line 106, in pandas._libs.index.IndexEngine.get_value
File "pandas/_libs/index.pyx", line 114, in pandas._libs.index.IndexEngine.get_value
File "pandas/_libs/index.pyx", line 164, in pandas._libs.index.IndexEngine.get_loc
KeyError: ('txt', 'occurred at index pm')

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "fullscript.py", line 748, in
partitiondata(tmpdata2)
File "fullscript.py", line 160, in partitiondata
data = procdata(tmp)
File "fullscript.py", line 677, in procdata
results = dd.from_pandas(dt,npartitions=nCores).map_partitions(lambda df1 : df1.apply(lambda x : proctxt(x['txt'], x['pm'], nlpe, nlpf))).compute(scheduler='distributed')
File "/usr/local/lib/python3.5/dist-packages/dask/dataframe/core.py", line 543, in map_partitions
return map_partitions(func, self, *args, **kwargs)
File "/usr/local/lib/python3.5/dist-packages/dask/dataframe/core.py", line 3640, in map_partitions
meta = _emulate(func, *args, udf=True, **kwargs2)
File "/usr/local/lib/python3.5/dist-packages/dask/dataframe/core.py", line 3601, in _emulate
return func(*_extract_meta(args, True), **_extract_meta(kwargs, True))
File "/usr/lib/python3.5/contextlib.py", line 77, in __exit__
self.gen.throw(type, value, traceback)
File "/usr/local/lib/python3.5/dist-packages/dask/dataframe/utils.py", line 154, in raise_on_meta_error
raise ValueError(msg)
ValueError: Metadata inference failed in `lambda`.

You have supplied a custom function and Dask is unable to
determine the type of output that that function returns.

To resolve this please provide a meta= keyword.
The docstring of the Dask function you ran should have more information.

Original error is below:
------------------------
KeyError('txt', 'occurred at index pm')

Traceback:
---------
File "/usr/local/lib/python3.5/dist-packages/dask/dataframe/utils.py", line 137, in raise_on_meta_error
yield
File "/usr/local/lib/python3.5/dist-packages/dask/dataframe/core.py", line 3601, in _emulate
return func(*_extract_meta(args, True), **_extract_meta(kwargs, True))
File "fullscript.py", line 677, in
results = dd.from_pandas(dt,npartitions=nCores).map_partitions(lambda df1 : df1.apply(lambda x : proctxt(x['txt'], x['pm'], nlpe, nlpf))).compute(scheduler='distributed')
File "/usr/local/lib/python3.5/dist-packages/pandas/core/frame.py", line 6014, in apply
return op.get_result()
File "/usr/local/lib/python3.5/dist-packages/pandas/core/apply.py", line 318, in get_result
return super(FrameRowApply, self).get_result()
File "/usr/local/lib/python3.5/dist-packages/pandas/core/apply.py", line 142, in get_result
return self.apply_standard()
File "/usr/local/lib/python3.5/dist-packages/pandas/core/apply.py", line 248, in apply_standard
self.apply_series_generator()
File "/usr/local/lib/python3.5/dist-packages/pandas/core/apply.py", line 277, in apply_series_generator
results[i] = self.f(v)
File "fullscript.py", line 677, in
results = dd.from_pandas(dt,npartitions=nCores).map_partitions(lambda df1 : df1.apply(lambda x : proctxt(x['txt'], x['pm'], nlpe, nlpf))).compute(scheduler='distributed')
File "/usr/local/lib/python3.5/dist-packages/pandas/core/series.py", line 767, in __getitem__
result = self.index.get_value(self, key)
File "/usr/local/lib/python3.5/dist-packages/pandas/core/indexes/base.py", line 3118, in get_value
tz=getattr(series.dtype, 'tz', None))
File "pandas/_libs/index.pyx", line 106, in pandas._libs.index.IndexEngine.get_value
File "pandas/_libs/index.pyx", line 114, in pandas._libs.index.IndexEngine.get_value
File "pandas/_libs/index.pyx", line 164, in pandas._libs.index.IndexEngine.get_loc
```

Thanks

Erico