We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
read_parquet not passing through categories
Although fastparquet correctly loads a categorical, and it shows up in the underlying pandas dataframe, this information does not appear in the dask dataframe.

Failing case:
```
def test_cats_in_parquet():
    s = pd.Series([b'a', b'b', b'c']*20, dtype='category')
    d = pd.DataFrame({'b': s})
    df = dd.from_pandas(d, npartitions=1)
    to_parquet('temp.parq', df, write_index=True)
    df2 = read_parquet('temp.parq', index='index', categories=['b'])
    assert df2.compute().b.cat,categories == df.b.cat.categories
    assert df2.b.cat.categories == df.b.cat.categories
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dask/dataframe/io/parquet.py`
Content:
```
1 import struct
2 
3 from toolz import first, partial
4 
5 from ..core import DataFrame, Series
6 from ..utils import make_meta
7 from ...base import compute, tokenize, normalize_token
8 from ...delayed import delayed
9 from ...bytes.core import OpenFileCreator
10 
11 try:
12     import fastparquet
13     from fastparquet import parquet_thrift
14     from fastparquet.core import read_row_group_file
15     default_encoding = parquet_thrift.Encoding.PLAIN
16 except:
17     fastparquet = False
18     default_encoding = None
19 
20 
21 def read_parquet(path, columns=None, filters=None, categories=None, index=None):
22     """
23     Read Dask DataFrame from ParquetFile
24 
25     This reads a directory of Parquet data into a Dask.dataframe, one file per
26     partition.  It selects the index among the sorted columns if any exist.
27 
28     Parameters
29     ----------
30     path : string
31         Source directory for data.
32         Prepend with protocol like ``s3://`` or ``hdfs://`` for remote data.
33     columns: list or None
34         List of column names to load
35     filters: list
36         List of filters to apply, like ``[('x', '>' 0), ...]``
37     index: string or None
38         Name of index column to use if that column is sorted
39     categories: list or None
40         For any fields listed here, if the parquet encoding is Dictionary,
41         the column will be created with dtype category. Use only if it is
42         guaranteed that the column is encoded as dictionary in all row-groups.
43 
44     Examples
45     --------
46     >>> df = read_parquet('s3://bucket/my-parquet-data')  # doctest: +SKIP
47 
48     See Also
49     --------
50     to_parquet
51     """
52     if fastparquet is False:
53         raise ImportError("fastparquet not installed")
54     if filters is None:
55         filters = []
56     myopen = OpenFileCreator(path, compression=None, text=False)
57 
58     if isinstance(columns, list):
59         columns = tuple(columns)
60 
61     try:
62         pf = fastparquet.ParquetFile(path + myopen.fs.sep + '_metadata',
63                                      open_with=myopen,
64                                      sep=myopen.fs.sep)
65     except:
66         pf = fastparquet.ParquetFile(path, open_with=myopen, sep=myopen.fs.sep)
67 
68     name = 'read-parquet-' + tokenize(pf, columns, categories)
69 
70     rgs = [rg for rg in pf.row_groups if
71            not(fastparquet.api.filter_out_stats(rg, filters, pf.helper)) and
72            not(fastparquet.api.filter_out_cats(rg, filters))]
73 
74     # Find an index among the partially sorted columns
75     minmax = fastparquet.api.sorted_partitioned_columns(pf)
76 
77     if index is False:
78         index_col = None
79     elif len(minmax) > 1:
80         if index:
81             index_col = index
82         else:
83             raise ValueError("Multiple possible indexes exist: %s.  "
84                              "Please select one with index='index-name'"
85                              % sorted(minmax))
86     elif len(minmax) == 1:
87         index_col = first(minmax)
88     else:
89         index_col = None
90 
91     if columns is None:
92         all_columns = tuple(pf.columns + list(pf.cats))
93     else:
94         all_columns = columns
95     if not isinstance(all_columns, tuple):
96         out_type = Series
97         all_columns = (all_columns,)
98     else:
99         out_type = DataFrame
100     if index_col and index_col not in all_columns:
101         all_columns = all_columns + (index_col,)
102 
103     # TODO: if categories vary from one rg to next, need to cope
104     dtypes = {k: ('category' if k in (categories or []) else v) for k, v in
105               pf.dtypes.items() if k in all_columns}
106 
107     meta = make_meta(dtypes)
108     if index_col:
109         meta = meta.set_index(index_col)
110 
111     if out_type == Series:
112         assert len(meta.columns) == 1
113         meta = meta[meta.columns[0]]
114 
115     dsk = {(name, i): (read_parquet_row_group, myopen, pf.row_group_filename(rg),
116                        index_col, all_columns, rg, out_type == Series,
117                        categories, pf.helper, pf.cats)
118            for i, rg in enumerate(rgs)}
119 
120     if index_col:
121         divisions = list(minmax[index_col]['min']) + [minmax[index_col]['max'][-1]]
122     else:
123         divisions = (None,) * (len(rgs) + 1)
124 
125     return out_type(dsk, name, meta, divisions)
126 
127 
128 def read_parquet_row_group(open, fn, index, columns, rg, series, *args):
129     if not isinstance(columns, (tuple, list)):
130         columns = (columns,)
131         series = True
132     if index and index not in columns:
133         columns = columns + type(columns)([index])
134     df = read_row_group_file(fn, rg, columns, *args, open=open)
135     if index:
136         df = df.set_index(index)
137 
138     if series:
139         return df[df.columns[0]]
140     else:
141         return df
142 
143 
144 def to_parquet(path, df, encoding=default_encoding, compression=None,
145                write_index=None):
146     """
147     Write Dask.dataframe to parquet
148 
149     Notes
150     -----
151     Each partition will be written to a separate file.
152 
153     Parameters
154     ----------
155     path : string
156         Destination directory for data.  Prepend with protocol like ``s3://``
157         or ``hdfs://`` for remote data.
158     df : Dask.dataframe
159     encoding : parquet_thrift.Encoding
160     compression : string or dict
161         Either a string like "SNAPPY" or a dictionary mapping column names to
162         compressors like ``{"name": "GZIP", "values": "SNAPPY"}``
163     write_index : boolean
164         Whether or not to write the index.  Defaults to True *if* divisions are
165         known.
166 
167     Examples
168     --------
169     >>> df = dd.read_csv(...)  # doctest: +SKIP
170     >>> to_parquet('/path/to/output/', df, compression='SNAPPY')  # doctest: +SKIP
171 
172     See Also
173     --------
174     read_parquet: Read parquet data to dask.dataframe
175     """
176     if fastparquet is False:
177         raise ImportError("fastparquet not installed")
178 
179     myopen = OpenFileCreator(path, compression=None, text=False)
180     myopen.fs.mkdirs(path)
181     sep = myopen.fs.sep
182     metadata_fn = sep.join([path, '_metadata'])
183 
184     if write_index is True or write_index is None and df.known_divisions:
185         df = df.reset_index()
186 
187     fmd = fastparquet.writer.make_metadata(df._meta_nonempty)
188 
189     partitions = df.to_delayed()
190     filenames = ['part.%i.parquet' % i for i in range(len(partitions))]
191     outfiles = [sep.join([path, fn]) for fn in filenames]
192 
193     writes = [delayed(fastparquet.writer.make_part_file)(
194               myopen(outfile, 'wb'), partition, fmd.schema,
195               compression=compression)
196               for outfile, partition in zip(outfiles, partitions)]
197 
198     out = compute(*writes)
199 
200     for fn, rg in zip(filenames, out):
201         for chunk in rg.columns:
202             chunk.file_path = fn
203         fmd.row_groups.append(rg)
204 
205     with myopen(metadata_fn, mode='wb') as f:
206         f.write(b'PAR1')
207         foot_size = fastparquet.writer.write_thrift(f, fmd)
208         f.write(struct.pack(b"<i", foot_size))
209         f.write(b'PAR1')
210         f.close()
211 
212     with myopen(sep.join([path, '_common_metadata']), mode='wb') as f:
213         fastparquet.writer.write_common_metadata(f, fmd)
214 
215 
216 if fastparquet:
217     @partial(normalize_token.register, fastparquet.ParquetFile)
218     def normalize_ParquetFile(pf):
219         return (type(pf), pf.fn, pf.sep) + normalize_token(pf.open)
220
```
--- 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/dask/dataframe/io/parquet.py b/dask/dataframe/io/parquet.py
--- a/dask/dataframe/io/parquet.py
+++ b/dask/dataframe/io/parquet.py
@@ -1,5 +1,6 @@
 import struct
 
+import pandas as pd
 from toolz import first, partial
 
 from ..core import DataFrame, Series
@@ -71,6 +72,11 @@
            not(fastparquet.api.filter_out_stats(rg, filters, pf.helper)) and
            not(fastparquet.api.filter_out_cats(rg, filters))]
 
+    # get category values from first row-group
+    categories = categories or []
+    cats = pf.grab_cats(categories)
+    categories = [cat for cat in categories if cats.get(cat, None) is not None]
+
     # Find an index among the partially sorted columns
     minmax = fastparquet.api.sorted_partitioned_columns(pf)
 
@@ -105,6 +111,9 @@
               pf.dtypes.items() if k in all_columns}
 
     meta = make_meta(dtypes)
+    for cat in categories:
+        meta[cat] = pd.Series(pd.Categorical([], categories=cats[cat]))
+
     if index_col:
         meta = meta.set_index(index_col)

{"golden_diff": "diff --git a/dask/dataframe/io/parquet.py b/dask/dataframe/io/parquet.py\n--- a/dask/dataframe/io/parquet.py\n+++ b/dask/dataframe/io/parquet.py\n@@ -1,5 +1,6 @@\n import struct\n \n+import pandas as pd\n from toolz import first, partial\n \n from ..core import DataFrame, Series\n@@ -71,6 +72,11 @@\n            not(fastparquet.api.filter_out_stats(rg, filters, pf.helper)) and\n            not(fastparquet.api.filter_out_cats(rg, filters))]\n \n+    # get category values from first row-group\n+    categories = categories or []\n+    cats = pf.grab_cats(categories)\n+    categories = [cat for cat in categories if cats.get(cat, None) is not None]\n+\n     # Find an index among the partially sorted columns\n     minmax = fastparquet.api.sorted_partitioned_columns(pf)\n \n@@ -105,6 +111,9 @@\n               pf.dtypes.items() if k in all_columns}\n \n     meta = make_meta(dtypes)\n+    for cat in categories:\n+        meta[cat] = pd.Series(pd.Categorical([], categories=cats[cat]))\n+\n     if index_col:\n         meta = meta.set_index(index_col)\n", "issue": "read_parquet not passing through categories\nAlthough fastparquet correctly loads a categorical, and it shows up in the underlying pandas dataframe, this information does not appear in the dask dataframe.\r\n\r\nFailing case:\r\n```\r\ndef test_cats_in_parquet():\r\n    s = pd.Series([b'a', b'b', b'c']*20, dtype='category')\r\n    d = pd.DataFrame({'b': s})\r\n    df = dd.from_pandas(d, npartitions=1)\r\n    to_parquet('temp.parq', df, write_index=True)\r\n    df2 = read_parquet('temp.parq', index='index', categories=['b'])\r\n    assert df2.compute().b.cat,categories == df.b.cat.categories\r\n    assert df2.b.cat.categories == df.b.cat.categories\r\n```\r\n\n", "before_files": [{"content": "import struct\n\nfrom toolz import first, partial\n\nfrom ..core import DataFrame, Series\nfrom ..utils import make_meta\nfrom ...base import compute, tokenize, normalize_token\nfrom ...delayed import delayed\nfrom ...bytes.core import OpenFileCreator\n\ntry:\n    import fastparquet\n    from fastparquet import parquet_thrift\n    from fastparquet.core import read_row_group_file\n    default_encoding = parquet_thrift.Encoding.PLAIN\nexcept:\n    fastparquet = False\n    default_encoding = None\n\n\ndef read_parquet(path, columns=None, filters=None, categories=None, index=None):\n    \"\"\"\n    Read Dask DataFrame from ParquetFile\n\n    This reads a directory of Parquet data into a Dask.dataframe, one file per\n    partition.  It selects the index among the sorted columns if any exist.\n\n    Parameters\n    ----------\n    path : string\n        Source directory for data.\n        Prepend with protocol like ``s3://`` or ``hdfs://`` for remote data.\n    columns: list or None\n        List of column names to load\n    filters: list\n        List of filters to apply, like ``[('x', '>' 0), ...]``\n    index: string or None\n        Name of index column to use if that column is sorted\n    categories: list or None\n        For any fields listed here, if the parquet encoding is Dictionary,\n        the column will be created with dtype category. Use only if it is\n        guaranteed that the column is encoded as dictionary in all row-groups.\n\n    Examples\n    --------\n    >>> df = read_parquet('s3://bucket/my-parquet-data')  # doctest: +SKIP\n\n    See Also\n    --------\n    to_parquet\n    \"\"\"\n    if fastparquet is False:\n        raise ImportError(\"fastparquet not installed\")\n    if filters is None:\n        filters = []\n    myopen = OpenFileCreator(path, compression=None, text=False)\n\n    if isinstance(columns, list):\n        columns = tuple(columns)\n\n    try:\n        pf = fastparquet.ParquetFile(path + myopen.fs.sep + '_metadata',\n                                     open_with=myopen,\n                                     sep=myopen.fs.sep)\n    except:\n        pf = fastparquet.ParquetFile(path, open_with=myopen, sep=myopen.fs.sep)\n\n    name = 'read-parquet-' + tokenize(pf, columns, categories)\n\n    rgs = [rg for rg in pf.row_groups if\n           not(fastparquet.api.filter_out_stats(rg, filters, pf.helper)) and\n           not(fastparquet.api.filter_out_cats(rg, filters))]\n\n    # Find an index among the partially sorted columns\n    minmax = fastparquet.api.sorted_partitioned_columns(pf)\n\n    if index is False:\n        index_col = None\n    elif len(minmax) > 1:\n        if index:\n            index_col = index\n        else:\n            raise ValueError(\"Multiple possible indexes exist: %s.  \"\n                             \"Please select one with index='index-name'\"\n                             % sorted(minmax))\n    elif len(minmax) == 1:\n        index_col = first(minmax)\n    else:\n        index_col = None\n\n    if columns is None:\n        all_columns = tuple(pf.columns + list(pf.cats))\n    else:\n        all_columns = columns\n    if not isinstance(all_columns, tuple):\n        out_type = Series\n        all_columns = (all_columns,)\n    else:\n        out_type = DataFrame\n    if index_col and index_col not in all_columns:\n        all_columns = all_columns + (index_col,)\n\n    # TODO: if categories vary from one rg to next, need to cope\n    dtypes = {k: ('category' if k in (categories or []) else v) for k, v in\n              pf.dtypes.items() if k in all_columns}\n\n    meta = make_meta(dtypes)\n    if index_col:\n        meta = meta.set_index(index_col)\n\n    if out_type == Series:\n        assert len(meta.columns) == 1\n        meta = meta[meta.columns[0]]\n\n    dsk = {(name, i): (read_parquet_row_group, myopen, pf.row_group_filename(rg),\n                       index_col, all_columns, rg, out_type == Series,\n                       categories, pf.helper, pf.cats)\n           for i, rg in enumerate(rgs)}\n\n    if index_col:\n        divisions = list(minmax[index_col]['min']) + [minmax[index_col]['max'][-1]]\n    else:\n        divisions = (None,) * (len(rgs) + 1)\n\n    return out_type(dsk, name, meta, divisions)\n\n\ndef read_parquet_row_group(open, fn, index, columns, rg, series, *args):\n    if not isinstance(columns, (tuple, list)):\n        columns = (columns,)\n        series = True\n    if index and index not in columns:\n        columns = columns + type(columns)([index])\n    df = read_row_group_file(fn, rg, columns, *args, open=open)\n    if index:\n        df = df.set_index(index)\n\n    if series:\n        return df[df.columns[0]]\n    else:\n        return df\n\n\ndef to_parquet(path, df, encoding=default_encoding, compression=None,\n               write_index=None):\n    \"\"\"\n    Write Dask.dataframe to parquet\n\n    Notes\n    -----\n    Each partition will be written to a separate file.\n\n    Parameters\n    ----------\n    path : string\n        Destination directory for data.  Prepend with protocol like ``s3://``\n        or ``hdfs://`` for remote data.\n    df : Dask.dataframe\n    encoding : parquet_thrift.Encoding\n    compression : string or dict\n        Either a string like \"SNAPPY\" or a dictionary mapping column names to\n        compressors like ``{\"name\": \"GZIP\", \"values\": \"SNAPPY\"}``\n    write_index : boolean\n        Whether or not to write the index.  Defaults to True *if* divisions are\n        known.\n\n    Examples\n    --------\n    >>> df = dd.read_csv(...)  # doctest: +SKIP\n    >>> to_parquet('/path/to/output/', df, compression='SNAPPY')  # doctest: +SKIP\n\n    See Also\n    --------\n    read_parquet: Read parquet data to dask.dataframe\n    \"\"\"\n    if fastparquet is False:\n        raise ImportError(\"fastparquet not installed\")\n\n    myopen = OpenFileCreator(path, compression=None, text=False)\n    myopen.fs.mkdirs(path)\n    sep = myopen.fs.sep\n    metadata_fn = sep.join([path, '_metadata'])\n\n    if write_index is True or write_index is None and df.known_divisions:\n        df = df.reset_index()\n\n    fmd = fastparquet.writer.make_metadata(df._meta_nonempty)\n\n    partitions = df.to_delayed()\n    filenames = ['part.%i.parquet' % i for i in range(len(partitions))]\n    outfiles = [sep.join([path, fn]) for fn in filenames]\n\n    writes = [delayed(fastparquet.writer.make_part_file)(\n              myopen(outfile, 'wb'), partition, fmd.schema,\n              compression=compression)\n              for outfile, partition in zip(outfiles, partitions)]\n\n    out = compute(*writes)\n\n    for fn, rg in zip(filenames, out):\n        for chunk in rg.columns:\n            chunk.file_path = fn\n        fmd.row_groups.append(rg)\n\n    with myopen(metadata_fn, mode='wb') as f:\n        f.write(b'PAR1')\n        foot_size = fastparquet.writer.write_thrift(f, fmd)\n        f.write(struct.pack(b\"<i\", foot_size))\n        f.write(b'PAR1')\n        f.close()\n\n    with myopen(sep.join([path, '_common_metadata']), mode='wb') as f:\n        fastparquet.writer.write_common_metadata(f, fmd)\n\n\nif fastparquet:\n    @partial(normalize_token.register, fastparquet.ParquetFile)\n    def normalize_ParquetFile(pf):\n        return (type(pf), pf.fn, pf.sep) + normalize_token(pf.open)\n", "path": "dask/dataframe/io/parquet.py"}], "after_files": [{"content": "import struct\n\nimport pandas as pd\nfrom toolz import first, partial\n\nfrom ..core import DataFrame, Series\nfrom ..utils import make_meta\nfrom ...base import compute, tokenize, normalize_token\nfrom ...delayed import delayed\nfrom ...bytes.core import OpenFileCreator\n\ntry:\n    import fastparquet\n    from fastparquet import parquet_thrift\n    from fastparquet.core import read_row_group_file\n    default_encoding = parquet_thrift.Encoding.PLAIN\nexcept:\n    fastparquet = False\n    default_encoding = None\n\n\ndef read_parquet(path, columns=None, filters=None, categories=None, index=None):\n    \"\"\"\n    Read Dask DataFrame from ParquetFile\n\n    This reads a directory of Parquet data into a Dask.dataframe, one file per\n    partition.  It selects the index among the sorted columns if any exist.\n\n    Parameters\n    ----------\n    path : string\n        Source directory for data.\n        Prepend with protocol like ``s3://`` or ``hdfs://`` for remote data.\n    columns: list or None\n        List of column names to load\n    filters: list\n        List of filters to apply, like ``[('x', '>' 0), ...]``\n    index: string or None\n        Name of index column to use if that column is sorted\n    categories: list or None\n        For any fields listed here, if the parquet encoding is Dictionary,\n        the column will be created with dtype category. Use only if it is\n        guaranteed that the column is encoded as dictionary in all row-groups.\n\n    Examples\n    --------\n    >>> df = read_parquet('s3://bucket/my-parquet-data')  # doctest: +SKIP\n\n    See Also\n    --------\n    to_parquet\n    \"\"\"\n    if fastparquet is False:\n        raise ImportError(\"fastparquet not installed\")\n    if filters is None:\n        filters = []\n    myopen = OpenFileCreator(path, compression=None, text=False)\n\n    if isinstance(columns, list):\n        columns = tuple(columns)\n\n    try:\n        pf = fastparquet.ParquetFile(path + myopen.fs.sep + '_metadata',\n                                     open_with=myopen,\n                                     sep=myopen.fs.sep)\n    except:\n        pf = fastparquet.ParquetFile(path, open_with=myopen, sep=myopen.fs.sep)\n\n    name = 'read-parquet-' + tokenize(pf, columns, categories)\n\n    rgs = [rg for rg in pf.row_groups if\n           not(fastparquet.api.filter_out_stats(rg, filters, pf.helper)) and\n           not(fastparquet.api.filter_out_cats(rg, filters))]\n\n    # get category values from first row-group\n    categories = categories or []\n    cats = pf.grab_cats(categories)\n    categories = [cat for cat in categories if cats.get(cat, None) is not None]\n\n    # Find an index among the partially sorted columns\n    minmax = fastparquet.api.sorted_partitioned_columns(pf)\n\n    if index is False:\n        index_col = None\n    elif len(minmax) > 1:\n        if index:\n            index_col = index\n        else:\n            raise ValueError(\"Multiple possible indexes exist: %s.  \"\n                             \"Please select one with index='index-name'\"\n                             % sorted(minmax))\n    elif len(minmax) == 1:\n        index_col = first(minmax)\n    else:\n        index_col = None\n\n    if columns is None:\n        all_columns = tuple(pf.columns + list(pf.cats))\n    else:\n        all_columns = columns\n    if not isinstance(all_columns, tuple):\n        out_type = Series\n        all_columns = (all_columns,)\n    else:\n        out_type = DataFrame\n    if index_col and index_col not in all_columns:\n        all_columns = all_columns + (index_col,)\n\n    # TODO: if categories vary from one rg to next, need to cope\n    dtypes = {k: ('category' if k in (categories or []) else v) for k, v in\n              pf.dtypes.items() if k in all_columns}\n\n    meta = make_meta(dtypes)\n    for cat in categories:\n        meta[cat] = pd.Series(pd.Categorical([], categories=cats[cat]))\n\n    if index_col:\n        meta = meta.set_index(index_col)\n\n    if out_type == Series:\n        assert len(meta.columns) == 1\n        meta = meta[meta.columns[0]]\n\n    dsk = {(name, i): (read_parquet_row_group, myopen, pf.row_group_filename(rg),\n                       index_col, all_columns, rg, out_type == Series,\n                       categories, pf.helper, pf.cats)\n           for i, rg in enumerate(rgs)}\n\n    if index_col:\n        divisions = list(minmax[index_col]['min']) + [minmax[index_col]['max'][-1]]\n    else:\n        divisions = (None,) * (len(rgs) + 1)\n\n    return out_type(dsk, name, meta, divisions)\n\n\ndef read_parquet_row_group(open, fn, index, columns, rg, series, *args):\n    if not isinstance(columns, (tuple, list)):\n        columns = (columns,)\n        series = True\n    if index and index not in columns:\n        columns = columns + type(columns)([index])\n    df = read_row_group_file(fn, rg, columns, *args, open=open)\n    if index:\n        df = df.set_index(index)\n\n    if series:\n        return df[df.columns[0]]\n    else:\n        return df\n\n\ndef to_parquet(path, df, encoding=default_encoding, compression=None,\n               write_index=None):\n    \"\"\"\n    Write Dask.dataframe to parquet\n\n    Notes\n    -----\n    Each partition will be written to a separate file.\n\n    Parameters\n    ----------\n    path : string\n        Destination directory for data.  Prepend with protocol like ``s3://``\n        or ``hdfs://`` for remote data.\n    df : Dask.dataframe\n    encoding : parquet_thrift.Encoding\n    compression : string or dict\n        Either a string like \"SNAPPY\" or a dictionary mapping column names to\n        compressors like ``{\"name\": \"GZIP\", \"values\": \"SNAPPY\"}``\n    write_index : boolean\n        Whether or not to write the index.  Defaults to True *if* divisions are\n        known.\n\n    Examples\n    --------\n    >>> df = dd.read_csv(...)  # doctest: +SKIP\n    >>> to_parquet('/path/to/output/', df, compression='SNAPPY')  # doctest: +SKIP\n\n    See Also\n    --------\n    read_parquet: Read parquet data to dask.dataframe\n    \"\"\"\n    if fastparquet is False:\n        raise ImportError(\"fastparquet not installed\")\n\n    myopen = OpenFileCreator(path, compression=None, text=False)\n    myopen.fs.mkdirs(path)\n    sep = myopen.fs.sep\n    metadata_fn = sep.join([path, '_metadata'])\n\n    if write_index is True or write_index is None and df.known_divisions:\n        df = df.reset_index()\n\n    fmd = fastparquet.writer.make_metadata(df._meta_nonempty)\n\n    partitions = df.to_delayed()\n    filenames = ['part.%i.parquet' % i for i in range(len(partitions))]\n    outfiles = [sep.join([path, fn]) for fn in filenames]\n\n    writes = [delayed(fastparquet.writer.make_part_file)(\n              myopen(outfile, 'wb'), partition, fmd.schema,\n              compression=compression)\n              for outfile, partition in zip(outfiles, partitions)]\n\n    out = compute(*writes)\n\n    for fn, rg in zip(filenames, out):\n        for chunk in rg.columns:\n            chunk.file_path = fn\n        fmd.row_groups.append(rg)\n\n    with myopen(metadata_fn, mode='wb') as f:\n        f.write(b'PAR1')\n        foot_size = fastparquet.writer.write_thrift(f, fmd)\n        f.write(struct.pack(b\"<i\", foot_size))\n        f.write(b'PAR1')\n        f.close()\n\n    with myopen(sep.join([path, '_common_metadata']), mode='wb') as f:\n        fastparquet.writer.write_common_metadata(f, fmd)\n\n\nif fastparquet:\n    @partial(normalize_token.register, fastparquet.ParquetFile)\n    def normalize_ParquetFile(pf):\n        return (type(pf), pf.fn, pf.sep) + normalize_token(pf.open)\n", "path": "dask/dataframe/io/parquet.py"}]}