E           File /usr/local/lib/python3.8/dist-packages/merlin/dag/executors.py:287, in DaskExecutor.transform(self, dataset, graph, output_dtypes, additional_columns, capture_dtypes, strict)
E               283 nodes = self._executor._output_nodes(graph)
E               285 self._clear_worker_cache()
E           --> 287 ddf = dataset.to_ddf()
E               289 # Check if we are only selecting columns (no transforms).
E               290 # If so, we should perform column selection at the ddf level.
E               291 # Otherwise, Dask will not push the column selection into the
E               292 # IO function.
E               293 if not nodes:
E           
E           File /usr/local/lib/python3.8/dist-packages/dask/dataframe/core.py:4686, in DataFrame.__getattr__(self, key)
E              4684     object.__getattribute__(self, key)
E              4685 else:
E           -> 4686     raise AttributeError("'DataFrame' object has no attribute %r" % key)
E           
E           AttributeError: 'DataFrame' object has no attribute 'to_ddf'