We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[FEA] Add DistributedClient API and use global client objects in Workflow and Dataset
**Is your feature request related to a problem? Please describe.**
In order to perform distributed ETL and/or spill data between device/host/disk, NVTabular currently requires the user to provide a Dask-Distributed (or Dask-CUDA) cluster.  Furthermore, the `nvt.Workflow` (and sometimes `nvt.Dataset`) need to be defined with an explicit `client=` argument in order for the distributed cluster to be used.  Although I feel strongly that it would be dangerous and messy for NVTabular to automatically spin up a distributed cluster by default, I do suspect that the user experience could be much better. 

**Describe the solution you'd like**
To improve the user experience of distributed ETL with NVTabular, I propose:

1. Simple `LocalCluster`/`LocalCUDACluster`-wrapper APIs be added to NVTabular so that users can enable multi-GPU processing and/or spilling without interacting with distributed/dask_cuda. I am not yet sure of the ideal API to expose in NVTabular, but perhaps something like `DistributedClient` (wrapping `distributed.Client`). This API could be used to automatically generate a local cluster (if the address of an existing cluster is not provided), and we could add a `cpu=False` kwarg to toggle between gpu and cpu mode.
2. [**DONE** in #1318] Automatically detect and **use** an existing Dask client object. NVTabular already [checks for a global dask client](https://github.com/NVIDIA-Merlin/NVTabular/blob/34d01d7e6090d6029ac40010ed79e1558f18759c/nvtabular/workflow/workflow.py#L88) in both `Workflow ` and `Dataset`.  However, these checks result in a UserWarning whenever a global client is detected.  Instead of warning the user, I propose that NVTabular automatically **use** the client object when it is detected (which is actually the defult behavior in `dask.dataframe` and `dask.array` anyway).

**Describe alternatives you've considered**
The alternative is to continue requiring the user to deploy Dask clusters/clients with `distributed`/`dask_cuda`, and require them to include an explicit `client` argument when defining an `Workflow` objects.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nvtabular/__init__.py`
Content:
```
1 #
2 # Copyright (c) 2022, NVIDIA CORPORATION.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 #
16 import warnings
17 
18 from merlin.graph import ColumnSelector
19 from merlin.schema import ColumnSchema, Schema
20 
21 from . import graph, io, workflow  # noqa
22 from ._version import get_versions
23 
24 # suppress some warnings with cudf warning about column ordering with dlpack
25 # and numba warning about deprecated environment variables
26 warnings.filterwarnings("ignore", module="cudf.io.dlpack")
27 warnings.filterwarnings("ignore", module="numba.cuda.envvars")
28 
29 
30 WorkflowNode = workflow.WorkflowNode
31 Workflow = workflow.Workflow
32 Dataset = io.dataset.Dataset
33 
34 
35 # Provides an alias of ColumnSelector so that old usages of ColumnGroup to
36 # select columns at the beginning of an operator chain don't break
37 def ColumnGroup(columns):
38     warnings.warn("ColumnGroup is deprecated, use ColumnSelector instead", DeprecationWarning)
39     return ColumnSelector(columns)
40 
41 
42 __all__ = [
43     "Workflow",
44     "Dataset",
45     "WorkflowNode",
46     "ColumnGroup",
47     "ColumnSelector",
48     "ColumnSchema",
49     "Schema",
50 ]
51 
52 # cudf warns about column ordering with dlpack methods, ignore it
53 warnings.filterwarnings("ignore", module="cudf.io.dlpack")
54 
55 
56 __version__ = get_versions()["version"]
57 del get_versions
58
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/nvtabular/__init__.py b/nvtabular/__init__.py
--- a/nvtabular/__init__.py
+++ b/nvtabular/__init__.py
@@ -18,7 +18,7 @@
 from merlin.graph import ColumnSelector
 from merlin.schema import ColumnSchema, Schema
 
-from . import graph, io, workflow  # noqa
+from . import graph, io, utils, workflow  # noqa
 from ._version import get_versions
 
 # suppress some warnings with cudf warning about column ordering with dlpack
@@ -30,6 +30,8 @@
 WorkflowNode = workflow.WorkflowNode
 Workflow = workflow.Workflow
 Dataset = io.dataset.Dataset
+Distributed = utils.Distributed
+Serial = utils.Serial
 
 
 # Provides an alias of ColumnSelector so that old usages of ColumnGroup to
@@ -47,6 +49,8 @@
     "ColumnSelector",
     "ColumnSchema",
     "Schema",
+    "Distributed",
+    "Serial",
 ]
 
 # cudf warns about column ordering with dlpack methods, ignore it

{"golden_diff": "diff --git a/nvtabular/__init__.py b/nvtabular/__init__.py\n--- a/nvtabular/__init__.py\n+++ b/nvtabular/__init__.py\n@@ -18,7 +18,7 @@\n from merlin.graph import ColumnSelector\n from merlin.schema import ColumnSchema, Schema\n \n-from . import graph, io, workflow  # noqa\n+from . import graph, io, utils, workflow  # noqa\n from ._version import get_versions\n \n # suppress some warnings with cudf warning about column ordering with dlpack\n@@ -30,6 +30,8 @@\n WorkflowNode = workflow.WorkflowNode\n Workflow = workflow.Workflow\n Dataset = io.dataset.Dataset\n+Distributed = utils.Distributed\n+Serial = utils.Serial\n \n \n # Provides an alias of ColumnSelector so that old usages of ColumnGroup to\n@@ -47,6 +49,8 @@\n     \"ColumnSelector\",\n     \"ColumnSchema\",\n     \"Schema\",\n+    \"Distributed\",\n+    \"Serial\",\n ]\n \n # cudf warns about column ordering with dlpack methods, ignore it\n", "issue": "[FEA] Add DistributedClient API and use global client objects in Workflow and Dataset\n**Is your feature request related to a problem? Please describe.**\r\nIn order to perform distributed ETL and/or spill data between device/host/disk, NVTabular currently requires the user to provide a Dask-Distributed (or Dask-CUDA) cluster.  Furthermore, the `nvt.Workflow` (and sometimes `nvt.Dataset`) need to be defined with an explicit `client=` argument in order for the distributed cluster to be used.  Although I feel strongly that it would be dangerous and messy for NVTabular to automatically spin up a distributed cluster by default, I do suspect that the user experience could be much better. \r\n\r\n**Describe the solution you'd like**\r\nTo improve the user experience of distributed ETL with NVTabular, I propose:\r\n\r\n1. Simple `LocalCluster`/`LocalCUDACluster`-wrapper APIs be added to NVTabular so that users can enable multi-GPU processing and/or spilling without interacting with distributed/dask_cuda. I am not yet sure of the ideal API to expose in NVTabular, but perhaps something like `DistributedClient` (wrapping `distributed.Client`). This API could be used to automatically generate a local cluster (if the address of an existing cluster is not provided), and we could add a `cpu=False` kwarg to toggle between gpu and cpu mode.\r\n2. [**DONE** in #1318] Automatically detect and **use** an existing Dask client object. NVTabular already [checks for a global dask client](https://github.com/NVIDIA-Merlin/NVTabular/blob/34d01d7e6090d6029ac40010ed79e1558f18759c/nvtabular/workflow/workflow.py#L88) in both `Workflow ` and `Dataset`.  However, these checks result in a UserWarning whenever a global client is detected.  Instead of warning the user, I propose that NVTabular automatically **use** the client object when it is detected (which is actually the defult behavior in `dask.dataframe` and `dask.array` anyway).\r\n\r\n**Describe alternatives you've considered**\r\nThe alternative is to continue requiring the user to deploy Dask clusters/clients with `distributed`/`dask_cuda`, and require them to include an explicit `client` argument when defining an `Workflow` objects.\r\n\n", "before_files": [{"content": "#\n# Copyright (c) 2022, NVIDIA CORPORATION.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\nimport warnings\n\nfrom merlin.graph import ColumnSelector\nfrom merlin.schema import ColumnSchema, Schema\n\nfrom . import graph, io, workflow  # noqa\nfrom ._version import get_versions\n\n# suppress some warnings with cudf warning about column ordering with dlpack\n# and numba warning about deprecated environment variables\nwarnings.filterwarnings(\"ignore\", module=\"cudf.io.dlpack\")\nwarnings.filterwarnings(\"ignore\", module=\"numba.cuda.envvars\")\n\n\nWorkflowNode = workflow.WorkflowNode\nWorkflow = workflow.Workflow\nDataset = io.dataset.Dataset\n\n\n# Provides an alias of ColumnSelector so that old usages of ColumnGroup to\n# select columns at the beginning of an operator chain don't break\ndef ColumnGroup(columns):\n    warnings.warn(\"ColumnGroup is deprecated, use ColumnSelector instead\", DeprecationWarning)\n    return ColumnSelector(columns)\n\n\n__all__ = [\n    \"Workflow\",\n    \"Dataset\",\n    \"WorkflowNode\",\n    \"ColumnGroup\",\n    \"ColumnSelector\",\n    \"ColumnSchema\",\n    \"Schema\",\n]\n\n# cudf warns about column ordering with dlpack methods, ignore it\nwarnings.filterwarnings(\"ignore\", module=\"cudf.io.dlpack\")\n\n\n__version__ = get_versions()[\"version\"]\ndel get_versions\n", "path": "nvtabular/__init__.py"}], "after_files": [{"content": "#\n# Copyright (c) 2022, NVIDIA CORPORATION.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\nimport warnings\n\nfrom merlin.graph import ColumnSelector\nfrom merlin.schema import ColumnSchema, Schema\n\nfrom . import graph, io, utils, workflow  # noqa\nfrom ._version import get_versions\n\n# suppress some warnings with cudf warning about column ordering with dlpack\n# and numba warning about deprecated environment variables\nwarnings.filterwarnings(\"ignore\", module=\"cudf.io.dlpack\")\nwarnings.filterwarnings(\"ignore\", module=\"numba.cuda.envvars\")\n\n\nWorkflowNode = workflow.WorkflowNode\nWorkflow = workflow.Workflow\nDataset = io.dataset.Dataset\nDistributed = utils.Distributed\nSerial = utils.Serial\n\n\n# Provides an alias of ColumnSelector so that old usages of ColumnGroup to\n# select columns at the beginning of an operator chain don't break\ndef ColumnGroup(columns):\n    warnings.warn(\"ColumnGroup is deprecated, use ColumnSelector instead\", DeprecationWarning)\n    return ColumnSelector(columns)\n\n\n__all__ = [\n    \"Workflow\",\n    \"Dataset\",\n    \"WorkflowNode\",\n    \"ColumnGroup\",\n    \"ColumnSelector\",\n    \"ColumnSchema\",\n    \"Schema\",\n    \"Distributed\",\n    \"Serial\",\n]\n\n# cudf warns about column ordering with dlpack methods, ignore it\nwarnings.filterwarnings(\"ignore\", module=\"cudf.io.dlpack\")\n\n\n__version__ = get_versions()[\"version\"]\ndel get_versions\n", "path": "nvtabular/__init__.py"}]}