Repository: dask/dask

Issue:
to_parquet partitioned by 'date'
I'm trying to save a dask dataframe `to_parquet` and partition it by date. The system won't allow the naming of the file `date=2022-01-01 00:00:00`. 
```
dates = pd.DataFrame({'date':pd.date_range(start='2022-01-01', end='2022-01-31', freq='D')})
ddata = dd.from_pandas(dates, npartitions=1)
ddata.to_parquet("dates.pq", partition_on="date")
```
```
OSError: [WinError 123] The filename, directory name, or volume label syntax is incorrect: '***/dask/dates.pq/date=2022-01-01 00:00:00'
```

Getting rid of the ` 00:00:00` would solve the issue. 

**Workaround**:
Save `date` as a string, but I would like it to be a date. 

I'm using dask '2023.5.1'

Discussion:
rjzamora: Thanks for the review on this @fjetter, and sorry for letting this slip through the cracks! I'll probably give this a last look tomorrow and merge if there are no follow-up comments.

Patch:
diff --git a/dask/dataframe/io/parquet/core.py b/dask/dataframe/io/parquet/core.py
index f00accde9..584279e73 100644
--- a/dask/dataframe/io/parquet/core.py
+++ b/dask/dataframe/io/parquet/core.py
@@ -317,10 +317,8 @@ def read_parquet(
         set to 'infer' or 'adaptive'. Default may be engine-dependant, but is
         128 MiB for the 'pyarrow' and 'fastparquet' engines.
     aggregate_files : bool or str, default None
-        WARNING: The ``aggregate_files`` argument will be deprecated in the future.
-        Please consider using ``from_map`` to create a DataFrame collection with a
-        custom file-to-partition mapping. If you strongly oppose the deprecation of
-        ``aggregate_files``, comment at https://github.com/dask/dask/issues/9051".
+        WARNING: Passing a string argument to ``aggregate_files`` will result
+        in experimental behavior. This behavior may change in the future.
 
         Whether distinct file paths may be aggregated into the same output
         partition. This parameter is only used when `split_row_groups` is set to
@@ -416,14 +414,11 @@ def read_parquet(
             FutureWarning,
         )
 
-    # "Pre-deprecation" warning for `aggregate_files`
-    if aggregate_files:
+    # FutureWarning for `aggregate_files`
+    if aggregate_files and isinstance(aggregate_files, str):
         warnings.warn(
-            "The `aggregate_files` argument will be deprecated in the future. "
-            "Please consider using `from_map` to create a DataFrame collection "
-            "with a custom file-to-partition mapping.\n\n"
-            "If you strongly oppose the deprecation of `aggregate_files`, "
-            "please comment at https://github.com/dask/dask/issues/9051",
+            "String support for `aggregate_files` is experimental. "
+            "Behavior may change in the future. ",
             FutureWarning,
         )
 
diff --git a/docs/source/dataframe-hive.rst b/docs/source/dataframe-hive.rst
new file mode 100644
index 000000000..8b8c2ba80
--- /dev/null
+++ b/docs/source/dataframe-hive.rst
@@ -0,0 +1,247 @@
+.. _dataframe.hive:
+
+Using Hive Partitioning with Dask
+=================================
+
+.. currentmodule:: dask.dataframe
+
+It is sometimes useful to write your dataset with a hive-like directory scheme.
+For example, if your dataframe contains ``'year'`` and ``'semester'`` columns,
+a hive-based directory structure might look something like the following::
+
+    output-path/
+    ├── year=2022/
+    │   ├── semester=fall/
+    │   │   └── part.0.parquet
+    │   └── semester=spring/
+    │       ├── part.0.parquet
+    │       └── part.1.parquet
+    └── year=2023/
+        └── semester=fall/
+            └── part.1.parquet
+
+The use of this self-describing structure implies that all rows within
+the ``'output-path/year=2022/semester=fall/'`` directory will contain
+the value ``2022`` in the ``'year'`` column and the value ``'fall'``
+in the ``'semester'`` column.
+
+The primary advantage of generating a hive-partitioned dataset
+is that certain IO filters can be applied by :func:`read_parquet`
+without the need to parse any file metadata. In other words,
+the following command will typically be faster when the dataset
+is already hive-partitioned on the ``'year'`` column.
+
+.. code-block:: python
+
+    >>> dd.read_parquet("output-path", filters=[("year", ">", 2022)])
+
+Writing Parquet Data with Hive Partitioning
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Dask's :func:`to_parquet` function will produce a hive-partitioned
+directory scheme automatically when the ``partition_on`` option is used.
+
+.. code-block:: python
+
+    >>> df.to_parquet("output-path", partition_on=["year", "semester"])
+
+    >>> os.listdir("output-path")
+    ["year=2022", "year=2023"]
+
+    >>> os.listdir("output-path/year=2022")
+    ["semester=fall", "semester=spring"]
+
+    >>> os.listdir("output-path/year=2022/semester=spring")
+    ['part.0.parquet', 'part.1.parquet']
+
+
+It is important to recognize that Dask will **not** aggregate the
+data files written within each of the leaf directories. This is
+because each of the DataFrame partitions is written independently 
+during the execution of the :func:`to_parquet` task graph. In order 
+to write out data for partition `i`, the partition-i write task will
+perform a `groupby` operation on columns ``["year", "semester"]``,
+and then each distinct group will be written to the corresponding
+directory using the file name ``'part.{i}.parquet'``. Therefore, it
+is possible for a hive-partitioned write to produce a large number
+of files in every leaf directory (one for each DataFrame partition).
+
+If your application requires you to produce a single parquet file
+for each hive partition, one possible solution is to sort or shuffle
+on the partitioning columns before calling :func:`to_parquet`.
+
+.. code-block:: python
+
+    >>> partition_on = ["year", "semester"]
+
+    >>> df.shuffle(on=partition_on).to_parquet(partition_on=partition_on)
+
+Using a global shuffle like this is extremely expensive, and should be
+avoided whenever possible. However, it is also guaranteed to produce
+the minimum number of files, which may be worth the sacrifice at times.
+
+
+Reading Parquet Data with Hive Partitioning
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+In most cases, :func:`read_parquet` will process hive-partitioned
+data automatically. By default, all hive-partitioned columns will
+be interpreted as categorical columns.
+
+.. code-block:: python
+
+    >>> ddf = dd.read_parquet("output-path", columns=["year", "semester"])
+
+    >>> ddf
+    Dask DataFrame Structure:
+                            year         semester
+    npartitions=4                                  
+                category[known]  category[known]
+                            ...              ...
+                            ...              ...
+                            ...              ...
+                            ...              ...
+    Dask Name: read-parquet, 1 graph layer
+
+    >>> ddf.compute()
+    year semester
+    0  2022     fall
+    1  2022     fall
+    2  2022     fall
+    3  2022   spring
+    4  2022   spring
+    5  2022   spring
+    6  2023     fall
+    7  2023     fall
+
+
+Defining a Custom Partitioning Schema
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+When utilizing ``engine='pyarrow'``, it is possible to specify a custom
+schema for the hive-partitioned columns. The columns will then be read
+using the specified types and not as `category`.
+
+.. code-block:: python
+
+    >>> schema = pa.schema([("year", pa.int16()), ("semester", pa.string())])
+
+    >>> ddf2 = dd.read_parquet(
+    ...     path,
+    ...     columns=["year", "semester"],
+    ...     dataset={"partitioning": {"flavor": "hive", "schema": schema}}
+    ... )
+    Dask DataFrame Structure:
+                    year semester
+    npartitions=4                
+                int16   object
+                    ...      ...
+                    ...      ...
+                    ...      ...
+                    ...      ...
+
+
+If any of your hive-partitioned columns contain null values, you
+**must** specify the partitioning schema in this way.
+
+Although it is not required, we also recommend that you specify
+the partitioning schema if you need to partition on high-cardinality
+columns. This is because the default ``'category'`` dtype will
+track the known categories in a way that can significantly increase
+the overall memory footprint of your Dask collection. In fact,
+:func:`read_parquet` already clears the "known categories" of other
+columns for this same reason (see :doc:`dataframe-categoricals`).
+
+
+Best Practices
+--------------
+
+Although hive partitioning can sometimes improve read performance
+by simplifying filtering, it can also lead to degraded performance
+and errors in other cases.
+
+
+Avoid High Cardinality
+~~~~~~~~~~~~~~~~~~~~~~
+
+A good rule of thumb is to avoid partitioning on `float` columns,
+or any column containing many unique values (i.e. high cardinality).
+
+The most common cause of poor user experience with hive partitioning
+is high-cardinality of the partitioning column(s). For example, if 
+you try to partition on a column with millions of unique values, then
+`:func:`to_parquet`` will need to generate millions of directories. The
+management of these directories is likely to put strain on the file
+system, and the need for many small files within each directory
+is sure to compound the issue.
+
+
+Use Simple Data Types for Partitioning
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Since hive-partitioned data is "self describing," we suggest that
+you avoid partitioning on complex data types, and opt for integer
+or string-based data types whenever possible. If your data type
+cannot be easily inferred from the string value used to define the
+directory name, then the IO engine may struggle to parse the values.
+
+For example, directly partitioning on a column with a ``datetime64``
+dtype might produce a directory name like the following::
+
+    output-path/
+    ├── date=2022-01-01 00:00:00/
+    ├── date=2022-02-01 00:00:00/
+    ├── ...
+    └── date=2022-12-01 00:00:00/
+
+These directory names will not be correctly interpreted as
+``datetime64`` values, and are even considered illegal on Windows
+systems. For more-reliable behavior, we recommend that such a column
+be decomposed into one or more "simple" columns. For example, one
+could easily use ``'date'`` to construct ``'year'``, ``'month'``,
+and ``'day'`` columns (as needed).
+
+
+Aggregate Files at Read Time
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. warning::
+    The ``aggregate_files`` argument is currently listed as
+    experimental. However, there are currently no plans to remove
+    the argument or change it's behavior in a future release.
+
+Since hive-partitioning will typically produce a large number of
+small files, :func:`read_parquet` performance will usually benefit
+from proper usage of the ``aggregate_files`` argument. Take the
+following dataset for example::
+
+    dataset-path/
+    ├── region=1/
+    │   ├── section=a/
+    │   │   └── 01.parquet
+    │   │   └── 02.parquet
+    │   │   └── 03.parquet
+    │   ├── section=b/
+    │   └── └── 04.parquet
+    │   └── └── 05.parquet
+    └── region=2/
+        ├── section=a/
+        │   ├── 06.parquet
+        │   ├── 07.parquet
+        │   ├── 08.parquet
+
+If we set ``aggregate_files=True`` for this case, we are telling Dask
+that any of the parquet data files may be aggregated into the same output
+DataFrame partition. If, instead, we specify the name of a partitioning
+column (e.g. ``'region'`` or ``'section'``), we allow the aggregation of
+any two files sharing a file path up to, and including, the corresponding
+directory name. For example, if ``aggregate_files`` is set to ``'section'``,
+``04.parquet`` and ``05.parquet`` may be aggregated together, but
+``03.parquet`` and ``04.parquet`` cannot be. If, however, ``aggregate_files``
+is set to ``'region'``, ``04.parquet`` may be aggregated with ``05.parquet``,
+**and** ``03.parquet`` may be aggregated with ``04.parquet``.
+
+Using ``aggregate_files`` will typically improve performance by making it
+more likely for DataFrame partitions to approach the size specified by the
+``blocksize`` argument. In contrast, default behavior may produce a large
+number of partitions that are much smaller than ``blocksize``.
diff --git a/docs/source/dataframe-parquet.rst b/docs/source/dataframe-parquet.rst
index e514269b5..813050ba9 100644
--- a/docs/source/dataframe-parquet.rst
+++ b/docs/source/dataframe-parquet.rst
@@ -260,6 +260,7 @@ moderate dataset sizes.
 File Names
 ~~~~~~~~~~
 
+Unless the `partition_on` option is used (see :doc:`dataframe-hive`),
 :func:`to_parquet` will write one file per Dask dataframe partition to the
 output directory. By default these files will have names like
 ``part.0.parquet``, ``part.1.parquet``, etc. If you wish to alter this naming
@@ -279,6 +280,15 @@ their partition indices.
     >>> os.listdir("/path/to/parquet")
     ["data-0.parquet", "data-1.parquet", "data-2.parquet"]
 
+Hive Partitioning
+~~~~~~~~~~~~~~~~~
+
+It is sometimes useful to write a parquet dataset with a hive-like directory scheme
+(e.g. ``'/year=2022/month=12/day=25'``). :func:`to_parquet` will automatically
+produce a dataset with this kind of directory structure when the ``partition_on``
+option is used. In most cases, :func:`to_parquet` will handle hive partitioning
+automatically. See :doc:`dataframe-hive` for more information.
+
 .. _parquet: https://parquet.apache.org/
 .. _glob string: https://docs.python.org/3/library/glob.html
 .. _fsspec: https://filesystem-spec.readthedocs.io
diff --git a/docs/source/dataframe.rst b/docs/source/dataframe.rst
index 6c3f45996..1278411c6 100644
--- a/docs/source/dataframe.rst
+++ b/docs/source/dataframe.rst
@@ -17,6 +17,7 @@ Dask DataFrame
    dataframe-categoricals.rst
    dataframe-extend.rst
    dataframe-parquet.rst
+   dataframe-hive.rst
    dataframe-sql.rst
    dataframe-api.rst
 


Test patch:
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 6c53cd6cb..a4ea31d11 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -2861,13 +2861,12 @@ def test_split_row_groups_int_aggregate_files(tmpdir, engine, split_row_groups):
 
     # Read back with both `split_row_groups>1` and
     # `aggregate_files=True`
-    with pytest.warns(FutureWarning, match="argument will be deprecated"):
-        ddf2 = dd.read_parquet(
-            str(tmpdir),
-            engine=engine,
-            split_row_groups=split_row_groups,
-            aggregate_files=True,
-        )
+    ddf2 = dd.read_parquet(
+        str(tmpdir),
+        engine=engine,
+        split_row_groups=split_row_groups,
+        aggregate_files=True,
+    )
 
     # Check that we are aggregating files as expected
     npartitions_expected = math.ceil((size / row_group_size) / split_row_groups)
@@ -3052,13 +3051,23 @@ def test_split_adaptive_files(
         write_index=False,
     )
 
-    with pytest.warns(FutureWarning, match="argument will be deprecated"):
+    aggregate_files = partition_on if partition_on else True
+    if isinstance(aggregate_files, str):
+        with pytest.warns(FutureWarning, match="Behavior may change"):
+            ddf2 = dd.read_parquet(
+                str(tmpdir),
+                engine=read_engine,
+                blocksize=blocksize,
+                split_row_groups="adaptive",
+                aggregate_files=aggregate_files,
+            )
+    else:
         ddf2 = dd.read_parquet(
             str(tmpdir),
             engine=read_engine,
             blocksize=blocksize,
             split_row_groups="adaptive",
-            aggregate_files=partition_on if partition_on else True,
+            aggregate_files=aggregate_files,
         )
 
     # Check that files where aggregated as expected
@@ -3103,7 +3112,7 @@ def test_split_adaptive_aggregate_files(
         partition_on=partition_on,
         write_index=False,
     )
-    with pytest.warns(FutureWarning, match="argument will be deprecated"):
+    with pytest.warns(FutureWarning, match="Behavior may change"):
         ddf2 = dd.read_parquet(
             str(tmpdir),
             engine=read_engine,
@@ -3157,16 +3166,15 @@ def test_split_adaptive_blocksize(tmpdir, blocksize, engine, metadata):
         assert "_metadata" not in files
         path = os.path.join(dirname, "*.parquet")
 
-    with pytest.warns(FutureWarning, match="argument will be deprecated"):
-        ddf2 = dd.read_parquet(
-            path,
-            engine=engine,
-            blocksize=blocksize,
-            split_row_groups="adaptive",
-            calculate_divisions=True,
-            index="index",
-            aggregate_files=True,
-        )
+    ddf2 = dd.read_parquet(
+        path,
+        engine=engine,
+        blocksize=blocksize,
+        split_row_groups="adaptive",
+        calculate_divisions=True,
+        index="index",
+        aggregate_files=True,
+    )
 
     assert_eq(ddf1, ddf2, check_divisions=False)