Repository: dask/dask
Pull request: Add documentation for hive partitioning
- Adds much-needed documentation on hive partitioning
- Moves ``aggregate_files`` from "pre-deprecation" to "experimental"
  - I seriously doubt we will want to remove the option any time soon.  However, I'm still not 100% happy with it yet.

Closes #10406


Problem statement:
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'

Hints from issue discussion:
Thanks for the report @marcdelabarrera!  I cannot reproduce on my end, so I'm thinking this is probably specific to windows. Can you confirm if you are using the "pyarrow" engine (by specifying `engine="pyarrow"`)?
Thank you for the reply. @rjzamora. 

I tried with `pyarrow` and I am getting the same error. I tried the same on Linxu and indeed worked. The problem is that on Windows, `:` is an invalid character to write in a directory name, while it does not seem to be the case with other OS.

I guess a partial fix is that the file was saved as `yyyy-mm-dd`, although this would drop information regarding hours, minutes, and seconds, or replacing `:` for some other allowed character.
>The problem is that on Windows, : is an invalid character to write in a directory name

Yes, you are definitely correct about this. I'll try to see if pyarrow's native `write_to_dataset` (or other systems) do something special to handle this problem.
>I guess a partial fix is that the file was saved as yyyy-mm-dd, although this would drop information regarding hours, minutes, and seconds, or replacing : for some other allowed character.

I'm not really finding any convenient ways to automatically handle the `":"` problem while partitioning by a `date_range` column. In practice, most hive partitioning tends to be on a specific "day", "month" or "year" (the second is not typically something you want to partition on). Therefore, if you need to track the time information somewhere, you may want to split your `"date"` column into distinct `"date"` and `"time"` columns, and only use the `"date"` column for partitioning.  If you don't need the time information, then doing the following before the write is probably your best bet:

```python
dates["date"] = dates["date"].astype("date64[pyarrow]")
```

Commit message: fix typo

Patch to review:
diff --git a/dask/dataframe/io/parquet/core.py b/dask/dataframe/io/parquet/core.py
index f00accde907..5db293f4470 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: The ``aggregate_files`` argument is experimental. 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`
+    # FutureWarning for `aggregate_files`
     if aggregate_files:
         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",
+            "The `aggregate_files` argument is experimental. "
+            "Behavior may change in the future. ",
             FutureWarning,
         )
 
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 6c53cd6cb92..f599b76f37e 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -2861,7 +2861,7 @@ 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"):
+    with pytest.warns(FutureWarning, match="Behavior may change"):
         ddf2 = dd.read_parquet(
             str(tmpdir),
             engine=engine,
@@ -3052,7 +3052,7 @@ def test_split_adaptive_files(
         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,
@@ -3103,7 +3103,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,7 +3157,7 @@ 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"):
+    with pytest.warns(FutureWarning, match="Behavior may change"):
         ddf2 = dd.read_parquet(
             path,
             engine=engine,
diff --git a/docs/source/dataframe-hive.rst b/docs/source/dataframe-hive.rst
new file mode 100644
index 00000000000..3473819af32
--- /dev/null
+++ b/docs/source/dataframe-hive.rst
@@ -0,0 +1,227 @@
+.. _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-partitioned directory scheme 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 scheme 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. This means that the parquet files themselves can
+(optionally) exclude the ``'year'`` and ``'semester'`` columns entirely.
+
+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 footer 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).
+
+
+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.
+
+.. 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.
+
+
+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
+~~~~~~~~~~~~~~~~~~~~~~
+
+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. 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 no current 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 e514269b516..813050ba93c 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 6c3f4599671..1278411c633 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
 


Code review comments:
diff_hunk: @@ -0,0 +1,227 @@
+.. _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-partitioned directory scheme 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 scheme 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. This means that the parquet files themselves can
+(optionally) exclude the ``'year'`` and ``'semester'`` columns entirely.
+
+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 footer metadata. In other words,
path: docs/source/dataframe-hive.rst
text: @user1:
I guess the typical user does not know about the internal structure of a parquet file, i.e they don't know what a footer is

```suggestion
without the need to parse any parquet metadata. In other words,
```

diff_hunk: @@ -0,0 +1,227 @@
+.. _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-partitioned directory scheme 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 scheme 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. This means that the parquet files themselves can
+(optionally) exclude the ``'year'`` and ``'semester'`` columns entirely.
path: docs/source/dataframe-hive.rst
text: @user1:
Is dask ignoring these or not? How can this be controlled and why would I care?

-> I think most users don't really care about this. This is only relevant if you are trying to write datasests with compat to spark in mind and I don't even know how they are doing it.



@author:
Sure, we can just remove this last sentence.

I think I was just trying to mention a key feature that can make manual IO code a bit tricky to write for partitioned data, but that discussion is probably out of scope here.