prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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.
  
  
  </issue>
  <code>
  [start of nvtabular/__init__.py]
  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 of nvtabular/__init__.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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
  
verification_info: {"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"}]}