We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[FEA] Warn on parquet row group sizes out of recommended bounds
**Is your feature request related to a problem? Please describe.**
We should warn on parquet files that contain row groups bigger than recommended - with actionable links and information for our customers


--- 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/io/parquet.py`
Content:
```
1 #
2 # Copyright (c) 2020, 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 functools
17 import logging
18 import os
19 import threading
20 import warnings
21 from io import BytesIO
22 from uuid import uuid4
23 
24 import cudf
25 import dask_cudf
26 from cudf.io.parquet import ParquetWriter as pwriter
27 from dask.utils import natural_sort_key
28 from pyarrow import parquet as pq
29 
30 from .dataset_engine import DatasetEngine
31 from .shuffle import Shuffle, _shuffle_gdf
32 from .writer import ThreadedWriter
33 
34 LOG = logging.getLogger("nvtabular")
35 
36 
37 class ParquetDatasetEngine(DatasetEngine):
38     """ParquetDatasetEngine is a Dask-based version of cudf.read_parquet."""
39 
40     def __init__(
41         self,
42         paths,
43         part_size,
44         storage_options,
45         row_groups_per_part=None,
46         legacy=False,
47         batch_size=None,  # Ignored
48     ):
49         super().__init__(paths, part_size, storage_options)
50         if row_groups_per_part is None:
51             path0 = self._dataset.pieces[0].path
52             rg_byte_size_0 = _memory_usage(cudf.io.read_parquet(path0, row_groups=0, row_group=0))
53             row_groups_per_part = self.part_size / rg_byte_size_0
54             if row_groups_per_part < 1.0:
55                 warnings.warn(
56                     f"Row group size {rg_byte_size_0} is bigger than requested part_size "
57                     f"{self.part_size}"
58                 )
59                 row_groups_per_part = 1.0
60 
61         self.row_groups_per_part = int(row_groups_per_part)
62 
63         assert self.row_groups_per_part > 0
64 
65     @property
66     @functools.lru_cache(1)
67     def _dataset(self):
68         paths = self.paths
69         fs = self.fs
70         if len(paths) > 1:
71             # This is a list of files
72             dataset = pq.ParquetDataset(paths, filesystem=fs, validate_schema=False)
73         elif fs.isdir(paths[0]):
74             # This is a directory
75             dataset = pq.ParquetDataset(paths[0], filesystem=fs, validate_schema=False)
76         else:
77             # This is a single file
78             dataset = pq.ParquetDataset(paths[0], filesystem=fs)
79         return dataset
80 
81     @property
82     @functools.lru_cache(1)
83     def num_rows(self):
84         # TODO: Avoid parsing metadata here if we can confirm upstream dask
85         # can get the length efficiently (in all practical cases)
86         dataset = self._dataset
87         if dataset.metadata:
88             # We have a metadata file
89             return dataset.metadata.num_rows
90         else:
91             # Sum up row-group sizes manually
92             num_rows = 0
93             for piece in dataset.pieces:
94                 num_rows += piece.get_metadata().num_rows
95             return num_rows
96 
97     def to_ddf(self, columns=None):
98         return dask_cudf.read_parquet(
99             self.paths,
100             columns=columns,
101             # can't omit reading the index in if we aren't being passed columns
102             index=None if columns is None else False,
103             gather_statistics=False,
104             split_row_groups=self.row_groups_per_part,
105             storage_options=self.storage_options,
106         )
107 
108 
109 class ParquetWriter(ThreadedWriter):
110     def __init__(self, out_dir, **kwargs):
111         super().__init__(out_dir, **kwargs)
112         self.data_paths = []
113         self.data_writers = []
114         self.data_bios = []
115         self._lock = threading.RLock()
116 
117     def _get_filename(self, i):
118         if self.use_guid:
119             fn = f"{i}.{guid()}.parquet"
120         else:
121             fn = f"{i}.parquet"
122 
123         return os.path.join(self.out_dir, fn)
124 
125     def _get_or_create_writer(self, idx):
126         # lazily initializes a writer for the given index
127         with self._lock:
128             while len(self.data_writers) <= idx:
129                 path = self._get_filename(len(self.data_writers))
130                 self.data_paths.append(path)
131                 if self.bytes_io:
132                     bio = BytesIO()
133                     self.data_bios.append(bio)
134                     self.data_writers.append(pwriter(bio, compression=None))
135                 else:
136                     self.data_writers.append(pwriter(path, compression=None))
137 
138             return self.data_writers[idx]
139 
140     def _write_table(self, idx, data, has_list_column=False):
141         if has_list_column:
142             # currently cudf doesn't support chunked parquet writers with list columns
143             # write out a new file, rather than stream multiple chunks to a single file
144             filename = self._get_filename(len(self.data_paths))
145             data.to_parquet(filename)
146             self.data_paths.append(filename)
147         else:
148             writer = self._get_or_create_writer(idx)
149             writer.write_table(data)
150 
151     def _write_thread(self):
152         while True:
153             item = self.queue.get()
154             try:
155                 if item is self._eod:
156                     break
157                 idx, data = item
158                 with self.write_locks[idx]:
159                     self._write_table(idx, data, False)
160             finally:
161                 self.queue.task_done()
162 
163     @classmethod
164     def write_special_metadata(cls, md, fs, out_dir):
165         # Sort metadata by file name and convert list of
166         # tuples to a list of metadata byte-blobs
167         md_list = [m[1] for m in sorted(list(md.items()), key=lambda x: natural_sort_key(x[0]))]
168 
169         # Aggregate metadata and write _metadata file
170         _write_pq_metadata_file(md_list, fs, out_dir)
171 
172     def _close_writers(self):
173         md_dict = {}
174         for writer, path in zip(self.data_writers, self.data_paths):
175             fn = path.split(self.fs.sep)[-1]
176             md_dict[fn] = writer.close(metadata_file_path=fn)
177         return md_dict
178 
179     def _bytesio_to_disk(self):
180         for bio, path in zip(self.data_bios, self.data_paths):
181             gdf = cudf.io.read_parquet(bio, index=False)
182             bio.close()
183             if self.shuffle == Shuffle.PER_WORKER:
184                 gdf = _shuffle_gdf(gdf)
185             gdf.to_parquet(path, compression=None, index=False)
186         return
187 
188 
189 def _write_pq_metadata_file(md_list, fs, path):
190     """ Converts list of parquet metadata objects into a single shared _metadata file. """
191     if md_list:
192         metadata_path = fs.sep.join([path, "_metadata"])
193         _meta = cudf.io.merge_parquet_filemetadata(md_list) if len(md_list) > 1 else md_list[0]
194         with fs.open(metadata_path, "wb") as fil:
195             _meta.tofile(fil)
196     return
197 
198 
199 def guid():
200     """Simple utility function to get random hex string"""
201     return uuid4().hex
202 
203 
204 def _memory_usage(df):
205     """this function is a workaround of a problem with getting memory usage of lists
206     in cudf0.16.  This can be deleted and just use `df.memory_usage(deep= True, index=True).sum()`
207     once we are using cudf 0.17 (fixed in https://github.com/rapidsai/cudf/pull/6549)"""
208     size = 0
209     for col in df._data.columns:
210         if cudf.utils.dtypes.is_list_dtype(col.dtype):
211             for child in col.base_children:
212                 size += child.__sizeof__()
213         else:
214             size += col._memory_usage(deep=True)
215     size += df.index.memory_usage(deep=True)
216     return size
217
```
--- 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/io/parquet.py b/nvtabular/io/parquet.py
--- a/nvtabular/io/parquet.py
+++ b/nvtabular/io/parquet.py
@@ -53,8 +53,12 @@
             row_groups_per_part = self.part_size / rg_byte_size_0
             if row_groups_per_part < 1.0:
                 warnings.warn(
-                    f"Row group size {rg_byte_size_0} is bigger than requested part_size "
-                    f"{self.part_size}"
+                    f"Row group memory size ({rg_byte_size_0}) (bytes) of parquet file is bigger"
+                    f" than requested part_size ({self.part_size}) for the NVTabular dataset."
+                    f"A row group memory size of 128 MB is generally recommended. You can find"
+                    f" info on how to set the row group size of parquet files in "
+                    f"https://nvidia.github.io/NVTabular/main/HowItWorks.html"
+                    f"#getting-your-data-ready-for-nvtabular"
                 )
                 row_groups_per_part = 1.0

{"golden_diff": "diff --git a/nvtabular/io/parquet.py b/nvtabular/io/parquet.py\n--- a/nvtabular/io/parquet.py\n+++ b/nvtabular/io/parquet.py\n@@ -53,8 +53,12 @@\n             row_groups_per_part = self.part_size / rg_byte_size_0\n             if row_groups_per_part < 1.0:\n                 warnings.warn(\n-                    f\"Row group size {rg_byte_size_0} is bigger than requested part_size \"\n-                    f\"{self.part_size}\"\n+                    f\"Row group memory size ({rg_byte_size_0}) (bytes) of parquet file is bigger\"\n+                    f\" than requested part_size ({self.part_size}) for the NVTabular dataset.\"\n+                    f\"A row group memory size of 128 MB is generally recommended. You can find\"\n+                    f\" info on how to set the row group size of parquet files in \"\n+                    f\"https://nvidia.github.io/NVTabular/main/HowItWorks.html\"\n+                    f\"#getting-your-data-ready-for-nvtabular\"\n                 )\n                 row_groups_per_part = 1.0\n", "issue": "[FEA] Warn on parquet row group sizes out of recommended bounds\n**Is your feature request related to a problem? Please describe.**\r\nWe should warn on parquet files that contain row groups bigger than recommended - with actionable links and information for our customers\r\n\n", "before_files": [{"content": "#\n# Copyright (c) 2020, 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 functools\nimport logging\nimport os\nimport threading\nimport warnings\nfrom io import BytesIO\nfrom uuid import uuid4\n\nimport cudf\nimport dask_cudf\nfrom cudf.io.parquet import ParquetWriter as pwriter\nfrom dask.utils import natural_sort_key\nfrom pyarrow import parquet as pq\n\nfrom .dataset_engine import DatasetEngine\nfrom .shuffle import Shuffle, _shuffle_gdf\nfrom .writer import ThreadedWriter\n\nLOG = logging.getLogger(\"nvtabular\")\n\n\nclass ParquetDatasetEngine(DatasetEngine):\n    \"\"\"ParquetDatasetEngine is a Dask-based version of cudf.read_parquet.\"\"\"\n\n    def __init__(\n        self,\n        paths,\n        part_size,\n        storage_options,\n        row_groups_per_part=None,\n        legacy=False,\n        batch_size=None,  # Ignored\n    ):\n        super().__init__(paths, part_size, storage_options)\n        if row_groups_per_part is None:\n            path0 = self._dataset.pieces[0].path\n            rg_byte_size_0 = _memory_usage(cudf.io.read_parquet(path0, row_groups=0, row_group=0))\n            row_groups_per_part = self.part_size / rg_byte_size_0\n            if row_groups_per_part < 1.0:\n                warnings.warn(\n                    f\"Row group size {rg_byte_size_0} is bigger than requested part_size \"\n                    f\"{self.part_size}\"\n                )\n                row_groups_per_part = 1.0\n\n        self.row_groups_per_part = int(row_groups_per_part)\n\n        assert self.row_groups_per_part > 0\n\n    @property\n    @functools.lru_cache(1)\n    def _dataset(self):\n        paths = self.paths\n        fs = self.fs\n        if len(paths) > 1:\n            # This is a list of files\n            dataset = pq.ParquetDataset(paths, filesystem=fs, validate_schema=False)\n        elif fs.isdir(paths[0]):\n            # This is a directory\n            dataset = pq.ParquetDataset(paths[0], filesystem=fs, validate_schema=False)\n        else:\n            # This is a single file\n            dataset = pq.ParquetDataset(paths[0], filesystem=fs)\n        return dataset\n\n    @property\n    @functools.lru_cache(1)\n    def num_rows(self):\n        # TODO: Avoid parsing metadata here if we can confirm upstream dask\n        # can get the length efficiently (in all practical cases)\n        dataset = self._dataset\n        if dataset.metadata:\n            # We have a metadata file\n            return dataset.metadata.num_rows\n        else:\n            # Sum up row-group sizes manually\n            num_rows = 0\n            for piece in dataset.pieces:\n                num_rows += piece.get_metadata().num_rows\n            return num_rows\n\n    def to_ddf(self, columns=None):\n        return dask_cudf.read_parquet(\n            self.paths,\n            columns=columns,\n            # can't omit reading the index in if we aren't being passed columns\n            index=None if columns is None else False,\n            gather_statistics=False,\n            split_row_groups=self.row_groups_per_part,\n            storage_options=self.storage_options,\n        )\n\n\nclass ParquetWriter(ThreadedWriter):\n    def __init__(self, out_dir, **kwargs):\n        super().__init__(out_dir, **kwargs)\n        self.data_paths = []\n        self.data_writers = []\n        self.data_bios = []\n        self._lock = threading.RLock()\n\n    def _get_filename(self, i):\n        if self.use_guid:\n            fn = f\"{i}.{guid()}.parquet\"\n        else:\n            fn = f\"{i}.parquet\"\n\n        return os.path.join(self.out_dir, fn)\n\n    def _get_or_create_writer(self, idx):\n        # lazily initializes a writer for the given index\n        with self._lock:\n            while len(self.data_writers) <= idx:\n                path = self._get_filename(len(self.data_writers))\n                self.data_paths.append(path)\n                if self.bytes_io:\n                    bio = BytesIO()\n                    self.data_bios.append(bio)\n                    self.data_writers.append(pwriter(bio, compression=None))\n                else:\n                    self.data_writers.append(pwriter(path, compression=None))\n\n            return self.data_writers[idx]\n\n    def _write_table(self, idx, data, has_list_column=False):\n        if has_list_column:\n            # currently cudf doesn't support chunked parquet writers with list columns\n            # write out a new file, rather than stream multiple chunks to a single file\n            filename = self._get_filename(len(self.data_paths))\n            data.to_parquet(filename)\n            self.data_paths.append(filename)\n        else:\n            writer = self._get_or_create_writer(idx)\n            writer.write_table(data)\n\n    def _write_thread(self):\n        while True:\n            item = self.queue.get()\n            try:\n                if item is self._eod:\n                    break\n                idx, data = item\n                with self.write_locks[idx]:\n                    self._write_table(idx, data, False)\n            finally:\n                self.queue.task_done()\n\n    @classmethod\n    def write_special_metadata(cls, md, fs, out_dir):\n        # Sort metadata by file name and convert list of\n        # tuples to a list of metadata byte-blobs\n        md_list = [m[1] for m in sorted(list(md.items()), key=lambda x: natural_sort_key(x[0]))]\n\n        # Aggregate metadata and write _metadata file\n        _write_pq_metadata_file(md_list, fs, out_dir)\n\n    def _close_writers(self):\n        md_dict = {}\n        for writer, path in zip(self.data_writers, self.data_paths):\n            fn = path.split(self.fs.sep)[-1]\n            md_dict[fn] = writer.close(metadata_file_path=fn)\n        return md_dict\n\n    def _bytesio_to_disk(self):\n        for bio, path in zip(self.data_bios, self.data_paths):\n            gdf = cudf.io.read_parquet(bio, index=False)\n            bio.close()\n            if self.shuffle == Shuffle.PER_WORKER:\n                gdf = _shuffle_gdf(gdf)\n            gdf.to_parquet(path, compression=None, index=False)\n        return\n\n\ndef _write_pq_metadata_file(md_list, fs, path):\n    \"\"\" Converts list of parquet metadata objects into a single shared _metadata file. \"\"\"\n    if md_list:\n        metadata_path = fs.sep.join([path, \"_metadata\"])\n        _meta = cudf.io.merge_parquet_filemetadata(md_list) if len(md_list) > 1 else md_list[0]\n        with fs.open(metadata_path, \"wb\") as fil:\n            _meta.tofile(fil)\n    return\n\n\ndef guid():\n    \"\"\"Simple utility function to get random hex string\"\"\"\n    return uuid4().hex\n\n\ndef _memory_usage(df):\n    \"\"\"this function is a workaround of a problem with getting memory usage of lists\n    in cudf0.16.  This can be deleted and just use `df.memory_usage(deep= True, index=True).sum()`\n    once we are using cudf 0.17 (fixed in https://github.com/rapidsai/cudf/pull/6549)\"\"\"\n    size = 0\n    for col in df._data.columns:\n        if cudf.utils.dtypes.is_list_dtype(col.dtype):\n            for child in col.base_children:\n                size += child.__sizeof__()\n        else:\n            size += col._memory_usage(deep=True)\n    size += df.index.memory_usage(deep=True)\n    return size\n", "path": "nvtabular/io/parquet.py"}], "after_files": [{"content": "#\n# Copyright (c) 2020, 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 functools\nimport logging\nimport os\nimport threading\nimport warnings\nfrom io import BytesIO\nfrom uuid import uuid4\n\nimport cudf\nimport dask_cudf\nfrom cudf.io.parquet import ParquetWriter as pwriter\nfrom dask.utils import natural_sort_key\nfrom pyarrow import parquet as pq\n\nfrom .dataset_engine import DatasetEngine\nfrom .shuffle import Shuffle, _shuffle_gdf\nfrom .writer import ThreadedWriter\n\nLOG = logging.getLogger(\"nvtabular\")\n\n\nclass ParquetDatasetEngine(DatasetEngine):\n    \"\"\"ParquetDatasetEngine is a Dask-based version of cudf.read_parquet.\"\"\"\n\n    def __init__(\n        self,\n        paths,\n        part_size,\n        storage_options,\n        row_groups_per_part=None,\n        legacy=False,\n        batch_size=None,  # Ignored\n    ):\n        super().__init__(paths, part_size, storage_options)\n        if row_groups_per_part is None:\n            path0 = self._dataset.pieces[0].path\n            rg_byte_size_0 = _memory_usage(cudf.io.read_parquet(path0, row_groups=0, row_group=0))\n            row_groups_per_part = self.part_size / rg_byte_size_0\n            if row_groups_per_part < 1.0:\n                warnings.warn(\n                    f\"Row group memory size ({rg_byte_size_0}) (bytes) of parquet file is bigger\"\n                    f\" than requested part_size ({self.part_size}) for the NVTabular dataset.\"\n                    f\"A row group memory size of 128 MB is generally recommended. You can find\"\n                    f\" info on how to set the row group size of parquet files in \"\n                    f\"https://nvidia.github.io/NVTabular/main/HowItWorks.html\"\n                    f\"#getting-your-data-ready-for-nvtabular\"\n                )\n                row_groups_per_part = 1.0\n\n        self.row_groups_per_part = int(row_groups_per_part)\n\n        assert self.row_groups_per_part > 0\n\n    @property\n    @functools.lru_cache(1)\n    def _dataset(self):\n        paths = self.paths\n        fs = self.fs\n        if len(paths) > 1:\n            # This is a list of files\n            dataset = pq.ParquetDataset(paths, filesystem=fs, validate_schema=False)\n        elif fs.isdir(paths[0]):\n            # This is a directory\n            dataset = pq.ParquetDataset(paths[0], filesystem=fs, validate_schema=False)\n        else:\n            # This is a single file\n            dataset = pq.ParquetDataset(paths[0], filesystem=fs)\n        return dataset\n\n    @property\n    @functools.lru_cache(1)\n    def num_rows(self):\n        # TODO: Avoid parsing metadata here if we can confirm upstream dask\n        # can get the length efficiently (in all practical cases)\n        dataset = self._dataset\n        if dataset.metadata:\n            # We have a metadata file\n            return dataset.metadata.num_rows\n        else:\n            # Sum up row-group sizes manually\n            num_rows = 0\n            for piece in dataset.pieces:\n                num_rows += piece.get_metadata().num_rows\n            return num_rows\n\n    def to_ddf(self, columns=None):\n        return dask_cudf.read_parquet(\n            self.paths,\n            columns=columns,\n            # can't omit reading the index in if we aren't being passed columns\n            index=None if columns is None else False,\n            gather_statistics=False,\n            split_row_groups=self.row_groups_per_part,\n            storage_options=self.storage_options,\n        )\n\n\nclass ParquetWriter(ThreadedWriter):\n    def __init__(self, out_dir, **kwargs):\n        super().__init__(out_dir, **kwargs)\n        self.data_paths = []\n        self.data_writers = []\n        self.data_bios = []\n        self._lock = threading.RLock()\n\n    def _get_filename(self, i):\n        if self.use_guid:\n            fn = f\"{i}.{guid()}.parquet\"\n        else:\n            fn = f\"{i}.parquet\"\n\n        return os.path.join(self.out_dir, fn)\n\n    def _get_or_create_writer(self, idx):\n        # lazily initializes a writer for the given index\n        with self._lock:\n            while len(self.data_writers) <= idx:\n                path = self._get_filename(len(self.data_writers))\n                self.data_paths.append(path)\n                if self.bytes_io:\n                    bio = BytesIO()\n                    self.data_bios.append(bio)\n                    self.data_writers.append(pwriter(bio, compression=None))\n                else:\n                    self.data_writers.append(pwriter(path, compression=None))\n\n            return self.data_writers[idx]\n\n    def _write_table(self, idx, data, has_list_column=False):\n        if has_list_column:\n            # currently cudf doesn't support chunked parquet writers with list columns\n            # write out a new file, rather than stream multiple chunks to a single file\n            filename = self._get_filename(len(self.data_paths))\n            data.to_parquet(filename)\n            self.data_paths.append(filename)\n        else:\n            writer = self._get_or_create_writer(idx)\n            writer.write_table(data)\n\n    def _write_thread(self):\n        while True:\n            item = self.queue.get()\n            try:\n                if item is self._eod:\n                    break\n                idx, data = item\n                with self.write_locks[idx]:\n                    self._write_table(idx, data, False)\n            finally:\n                self.queue.task_done()\n\n    @classmethod\n    def write_special_metadata(cls, md, fs, out_dir):\n        # Sort metadata by file name and convert list of\n        # tuples to a list of metadata byte-blobs\n        md_list = [m[1] for m in sorted(list(md.items()), key=lambda x: natural_sort_key(x[0]))]\n\n        # Aggregate metadata and write _metadata file\n        _write_pq_metadata_file(md_list, fs, out_dir)\n\n    def _close_writers(self):\n        md_dict = {}\n        for writer, path in zip(self.data_writers, self.data_paths):\n            fn = path.split(self.fs.sep)[-1]\n            md_dict[fn] = writer.close(metadata_file_path=fn)\n        return md_dict\n\n    def _bytesio_to_disk(self):\n        for bio, path in zip(self.data_bios, self.data_paths):\n            gdf = cudf.io.read_parquet(bio, index=False)\n            bio.close()\n            if self.shuffle == Shuffle.PER_WORKER:\n                gdf = _shuffle_gdf(gdf)\n            gdf.to_parquet(path, compression=None, index=False)\n        return\n\n\ndef _write_pq_metadata_file(md_list, fs, path):\n    \"\"\" Converts list of parquet metadata objects into a single shared _metadata file. \"\"\"\n    if md_list:\n        metadata_path = fs.sep.join([path, \"_metadata\"])\n        _meta = cudf.io.merge_parquet_filemetadata(md_list) if len(md_list) > 1 else md_list[0]\n        with fs.open(metadata_path, \"wb\") as fil:\n            _meta.tofile(fil)\n    return\n\n\ndef guid():\n    \"\"\"Simple utility function to get random hex string\"\"\"\n    return uuid4().hex\n\n\ndef _memory_usage(df):\n    \"\"\"this function is a workaround of a problem with getting memory usage of lists\n    in cudf0.16.  This can be deleted and just use `df.memory_usage(deep= True, index=True).sum()`\n    once we are using cudf 0.17 (fixed in https://github.com/rapidsai/cudf/pull/6549)\"\"\"\n    size = 0\n    for col in df._data.columns:\n        if cudf.utils.dtypes.is_list_dtype(col.dtype):\n            for child in col.base_children:\n                size += child.__sizeof__()\n        else:\n            size += col._memory_usage(deep=True)\n    size += df.index.memory_usage(deep=True)\n    return size\n", "path": "nvtabular/io/parquet.py"}]}