Repository: dask/dask

Issue:
[Documentation Request] Configuration  Reference
It would be really great if we had a page describing all of the possible configuration options. I am new to Dask. I am still figuring out the configuration options. I usually google around. It'd be great if that knowledge about the configuration settings, possible values, and explanations were centralized.  A great example of this is [Django's Settings Reference](https://docs.djangoproject.com/en/2.1/ref/settings/).

Thank you for listening.

Discussion:
mrocklin: I have two comments that I think we should consider:

## Hierarchy

I'm curious about how best to handle hierarchy.  Dask's configuration is nested, which I think helps with keeping things compartmentalized.  The current structure has one level of nesting, followed up with a flattened hierarchy.  This is a sensible choice, but there might be other choices.  

One approach would be to use subsections and subsubsections to handle the nesting, and then use a Sphinx TOC on top.  This would capture the hierarchical nature of our configuration and also maybe make it easy for users to navigate quickly to bits that they care about.  It also might be way more confusing.  I'm not sure.

## Single source of truth

We're now copying configuration into two places.  This might be necessary, but it's also something that we should be aware of and avoid if possible (although I have no suggestions on how best to avoid it).
mrocklin: One way to handle the single source of truth would be to have way more extensive comments in the config files themselves, and then autogenerate the docs.  This has tradeoffs though.
quasiben: @mrocklin, I briefly thought about subsections but was lazy -- as a test, I opted for the following:

> When defining nested configurations, the top level default value will be **blank**, with subsequent keys and values listed below

The single source of truth is important but I am -0.5 on adding more comments to the config.  With more text in the config it makes editing the YAML rather challenging and becomes error prone.  It's also worth noting that dask has configuration management in two locations:
- distributed.yaml
- dask.yaml

quasiben: I am looking into a smallish sphinx plugin + ruamel.yaml for loading comments
mrocklin: My hope is that using subsections wouldn't be terrible.  It might also help to avoid the `foo.bar.baz` formatting in the table (which is liable to get long).

Let's take a look at the scheduler section

```yaml

  scheduler:
    allowed-failures: 3     # number of retries before a task is considered bad
    bandwidth: 100000000    # 100 MB/s estimated worker-worker bandwidth
    blocked-handlers: []
    default-data-size: 1000
    # Number of seconds to wait until workers or clients are removed from the events log
    # after they have been removed from the scheduler
    events-cleanup-delay: 1h
    idle-timeout: null      # Shut down after this duration, like "1h" or "30 minutes"
    transition-log-length: 100000
    work-stealing: True     # workers should steal tasks from each other
    work-stealing-interval: 100ms  # Callback time for work stealing
    worker-ttl: null        # like '60s'. Time to live for workers.  They must heartbeat faster than this
    pickle: True            # Is the scheduler allowed to deserialize arbitrary bytestrings
    preload: []
    preload-argv: []
    unknown-task-duration: 500ms  # Default duration for all tasks with unknown durations ("15m", "2h")
    default-task-durations:  # How long we expect function names to run ("1h", "1s") (helps for long     tasks)
      rechunk-split: 1us
      shuffle-split: 1us
    validate: False         # Check scheduler state at every step for debugging
    dashboard:
      status:
        task-stream-length: 1000
      tasks:
        task-stream-length: 100000
      tls:
        ca-file: null
        key: null
        cert: null
    locks:
      lease-validation-interval: 10s  # The time to wait until an acquired semaphore is released if the  Client goes out of scope
```

So there are a lot of entries in the top level, which is great.  The table will be simple for them.  There are some larger sections, like scheduler.dashboard, which I think would make sense to break out to a subsection, and there are some smaller sections, like `locks` with a single element, that may not make sense to break out.  

I think that it probably won't be hard to make subsections (the RST for this is fortunately simple).  In some cases I think that we probably do want subsections, and in other cases the subsections might be small enough to even be a hindrance. I don't think that we want to mix approaches, but I'm not sure.

(I'm just thinking out loud here to get conversation going)
mrocklin: > The single source of truth is important but I am -0.5 on adding more comments to the config. With more text in the config it makes editing the YAML rather challenging and becomes error prone. 

Yeah, to be clear, I'm not pushing this.  I'm just dumping a bunch of options out there to spur some thinking.  This is going to be a lot of detailed work once we get something down.  I'd prefer that we only do that detailed work once.
quasiben: In #b06c46 I played with building an plugin to automate the table (just for fun).  It's not awful :)

Below is a screen shot
<img width="626" alt="Screen Shot 2020-04-06 at 5 53 22 PM" src="https://user-images.githubusercontent.com/1403768/78608914-ada09600-782f-11ea-9717-73ef9066c29f.png">

jcrist: Woot, this is nice to see @quasiben. I do agree with @mrocklin's concern about two sources of truth.

For dask client-side configuration, lately I've been ensuring the default file is heavily (and nicely) commented, and then directly including that in the docs. Examples:
- [dask-gateway](https://gateway.dask.org/configuration-user.html#default-configuration)
- [dask-yarn](https://yarn.dask.org/en/latest/configuration.html#default-configuration)

As another idea, the zero-to-jupyterhub project includes an extra validation script that tests that the `values.yaml` file matches the documented schema (see https://github.com/jupyterhub/zero-to-jupyterhub-k8s/blob/master/jupyterhub/validate.py). The schema is stored in json, which makes it easier to use it to generate a nice html table for the docs.

*I personally prefer the traitlets approach, but recognize that not everyone might prefer that approach, and moving from Dask's existing configuration system would be tricky. *
mrocklin: > *I personally prefer the traitlets approach, but recognize that not everyone might prefer that approach, and moving from Dask's existing configuration system would be tricky. *

Heh. That's me!  :)

Yeah, I personally would be sad with traitlets I think (although I probably should do more homework on it).  There is probably some happier medium though.  

> For dask client-side configuration, lately I've been ensuring the default file is heavily (and nicely) commented, and then directly including that in the docs. 

Those are nice.

> As another idea, the zero-to-jupyterhub project includes an extra validation script that tests that the values.yaml file matches the documented schema (see https://github.com/jupyterhub/zero-to-jupyterhub-k8s/blob/master/jupyterhub/validate.py). The schema is stored in json, which makes it easier to use it to generate a nice html table for the docs.

A companion file like this might not be bad, especially if adding to it could be done optionally and users didn't see it directly.
quasiben: > A companion file like this might not be bad, especially if adding to it could be done optionally and users didn't see it directly.

A companion file would need to duplicate the keys, no ? Maybe that's ok and it's a baby step in the direction of validation ?  
jcrist: > A companion file would need to duplicate the keys, no ? Maybe that's ok and it's a baby step in the direction of validation ?

To be clear, I think the value of the JupyterHub spec file approach is the validation script. So while the information is split between files (which could fall out of sync), there's a way to test that they remain in sync *and* that test could be run as part of our CI.
mrocklin: Here is an attempt with jsonschema (the thing that JuptyerHub uses)

https://github.com/dask/distributed/pull/3696

So far I'm decently happy with it.
quasiben: I spent a bit of time with @mrocklin this weekend using PR https://github.com/dask/distributed/pull/3696 to help with descriptions in the table.   We pushed on getting the table to render everything and while it works, the rendering is a bit off:

<img width="1069" alt="Screen Shot 2020-04-12 at 11 37 27 PM" src="https://user-images.githubusercontent.com/1403768/79090126-9faeb180-7d16-11ea-8632-2f42b26e828e.png">

I also spent some time adding a custom yaml to HTML styling.  This non-csv layout of the config perhaps balances verbose descriptions (which users will appreciate) without sacrifice layout/readability (which the csv-table does).  However, we lose fancy TOC linking:

<img width="686" alt="Screen Shot 2020-04-12 at 11 32 13 PM" src="https://user-images.githubusercontent.com/1403768/79090233-f6b48680-7d16-11ea-85f4-de4001250c22.png">

mrocklin: Woot

On Sun, Apr 12, 2020 at 8:40 PM Benjamin Zaitlen <notifications@github.com>
wrote:

> I spent a bit of time with @mrocklin <https://github.com/mrocklin> this
> weekend using PR dask/distributed#3696
> <https://github.com/dask/distributed/pull/3696> to help with descriptions
> in the table. We pushed on getting the table to render everything and while
> it works, the rendering is a bit off:
>
> [image: Screen Shot 2020-04-12 at 11 37 27 PM]
> <https://user-images.githubusercontent.com/1403768/79090126-9faeb180-7d16-11ea-8632-2f42b26e828e.png>
>
> I also spent some time adding a custom yaml to HTML styling. This non-csv
> layout of the config perhaps balances verbose descriptions (which users
> will appreciate) without sacrifice layout/readability (which the csv-table
> does). However, we lose fancy TOC linking:
>
> [image: Screen Shot 2020-04-12 at 11 32 13 PM]
> <https://user-images.githubusercontent.com/1403768/79090233-f6b48680-7d16-11ea-85f4-de4001250c22.png>
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/pull/6069#issuecomment-612735780>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AACKZTCGMTZ4LZJ6HDTO2FLRMKCUBANCNFSM4MCL7W4Q>
> .
>

jcrist: > This non-csv layout of the config perhaps balances verbose descriptions (which users will appreciate) without sacrifice layout/readability (which the csv-table does). However, we lose fancy TOC linking.

I personally prefer this style, and think you should be able to do TOC linking. Refs work fine for the setup I have for dask-gateway (https://gateway.dask.org/api-server.html), I feel like they should be doable here too.
quasiben: @jcrist this is ready for a review if you have some time.  

Note: i also included a dask-schema in the PR.  This made iterating a bit faster for me but also happy to move to another PR if you'd like.

Patch:
diff --git a/continuous_integration/environment-3.6.yaml b/continuous_integration/environment-3.6.yaml
index 7d6840ac4..ce2fdc801 100644
--- a/continuous_integration/environment-3.6.yaml
+++ b/continuous_integration/environment-3.6.yaml
@@ -12,6 +12,7 @@ dependencies:
   - moto
   - coverage
   - codecov
+  - jsonschema
   # Issue with conda package for NumPy 1.15?
   # See fastparquet-444
   # - fastparquet
diff --git a/continuous_integration/environment-3.7.yaml b/continuous_integration/environment-3.7.yaml
index 21b1b0dd9..6956dafe4 100644
--- a/continuous_integration/environment-3.7.yaml
+++ b/continuous_integration/environment-3.7.yaml
@@ -20,6 +20,7 @@ dependencies:
   - sqlalchemy
   - pyarrow>=0.14.0
   - coverage
+  - jsonschema
   # other -- IO
   - bcolz
   - blosc
diff --git a/continuous_integration/environment-3.8-dev.yaml b/continuous_integration/environment-3.8-dev.yaml
index 2d939caff..44f0a1706 100644
--- a/continuous_integration/environment-3.8-dev.yaml
+++ b/continuous_integration/environment-3.8-dev.yaml
@@ -19,6 +19,7 @@ dependencies:
   - tiledb-py
   - sqlalchemy
   - pyarrow>=0.14.0
+  - jsonschema
   # other -- IO
   - bcolz
   - blosc
diff --git a/continuous_integration/environment-3.8.yaml b/continuous_integration/environment-3.8.yaml
index d3fdc9117..fb9e520e8 100644
--- a/continuous_integration/environment-3.8.yaml
+++ b/continuous_integration/environment-3.8.yaml
@@ -20,6 +20,7 @@ dependencies:
   - sqlalchemy
   - pyarrow>=0.14.0
   - coverage
+  - jsonschema
   # other -- IO
   - bcolz
   - blosc
diff --git a/dask/array/core.py b/dask/array/core.py
index a91218c23..5d807fdcb 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -2751,6 +2751,12 @@ def from_array(
         raise ValueError(
             "Array is already a dask array. Use 'asarray' or " "'rechunk' instead."
         )
+    elif is_dask_collection(x):
+        warnings.warn(
+            "Passing an object to dask.array.from_array which is already a "
+            "Dask collection. This can lead to unexpected behavior."
+        )
+
     if isinstance(x, (list, tuple, memoryview) + np.ScalarType):
         x = np.array(x)
 
diff --git a/dask/dask-schema.yaml b/dask/dask-schema.yaml
new file mode 100644
index 000000000..db5af4a06
--- /dev/null
+++ b/dask/dask-schema.yaml
@@ -0,0 +1,72 @@
+properties:
+  temporary-directory:
+    type:
+      - string
+      - "null"
+    description: |
+      Temporary directory for local disk storage /tmp, /scratch,
+      or /local.  This directory is used during dask spill-to-disk operations.
+
+      When the value is "null" (default), dask will create a directory from
+      where dask was launched: `cwd/dask-worker-space`
+
+  dataframe:
+    type: object
+    properties:
+
+      shuffle-compression:
+        type:
+          - string
+          - "null"
+        description: |
+          Compression algorithm used for on disk-shuffling. Partd, the library used
+          for compression supports ZLib, BZ2, SNAPPY, and BLOSC
+
+  array:
+    type: object
+    properties:
+
+      svg:
+        type: object
+        properties:
+
+          size:
+            type: integer
+            description: |
+              The size of pixels used when displaying a dask array as an SVG image.
+              This is used, for example, for nice rendering in a Jupyter notebook
+
+  optimization:
+    type: object
+    properties:
+
+      fuse:
+        type: object
+        properties:
+        description: Options for Dask's task fusion optimizations
+
+          active:
+            type: boolean
+            description: |
+              Configuration setting to turn task fusion on/off
+
+          ave-width:
+            type: integer
+            description: |
+              Upper limit for width, where width = num_nodes / height, a good measure
+              of parallelizability
+
+          subraphs:
+            type: boolean
+            description: |
+              Configuration setting to control whether to fuse multiple tasks into
+              SubgraphCallable objects.
+
+          rename-keys:
+            type: boolean
+            description: |
+              Configuration setting to control renaming of the fused keys with
+              `default_fused_keys_renamer` or not. Renaming fused keys can keep the
+              graph more understandable and comprehensive, but it comes at the cost of
+              additional processing. If False, then the top-most key will be used. For
+              advanced usage, a function to create the new name is also accepted.
diff --git a/dask/dataframe/categorical.py b/dask/dataframe/categorical.py
index d07878a95..2eda088bf 100644
--- a/dask/dataframe/categorical.py
+++ b/dask/dataframe/categorical.py
@@ -11,6 +11,8 @@ from .utils import (
     is_scalar,
     is_categorical_dtype,
 )
+from . import methods
+from ..utils import Dispatch
 
 
 def _categorize_block(df, categories, index):
@@ -24,12 +26,16 @@ def _categorize_block(df, categories, index):
         if is_categorical_dtype(df[col]):
             df[col] = df[col].cat.set_categories(vals)
         else:
-            df[col] = pd.Categorical(df[col], categories=vals, ordered=False)
+            cat_dtype = categorical_dtype(meta=df[col], categories=vals, ordered=False)
+            df[col] = df[col].astype(cat_dtype)
     if index is not None:
         if is_categorical_dtype(df.index):
             ind = df.index.set_categories(index)
         else:
-            ind = pd.Categorical(df.index, categories=index, ordered=False)
+            cat_dtype = categorical_dtype(
+                meta=df.index, categories=index, ordered=False
+            )
+            ind = df.index.astype(dtype=cat_dtype)
         ind.name = df.index.name
         df.index = ind
     return df
@@ -57,7 +63,10 @@ def _get_categories_agg(parts):
         for k, v in p[0].items():
             res[k].append(v)
         res_ind.append(p[1])
-    res = {k: pd.concat(v, ignore_index=True).drop_duplicates() for k, v in res.items()}
+    res = {
+        k: methods.concat(v, ignore_index=True).drop_duplicates()
+        for k, v in res.items()
+    }
     if res_ind[0] is None:
         return res, None
     return res, res_ind[0].append(res_ind[1:]).drop_duplicates()
@@ -264,3 +273,16 @@ class CategoricalAccessor(Accessor):
             meta=meta,
             token="cat-set_categories",
         )
+
+
+categorical_dtype_dispatch = Dispatch("CategoricalDtype")
+
+
+def categorical_dtype(meta, categories=None, ordered=False):
+    func = categorical_dtype_dispatch.dispatch(type(meta))
+    return func(categories=categories, ordered=ordered)
+
+
+@categorical_dtype_dispatch.register((pd.DataFrame, pd.Series, pd.Index))
+def categorical_dtype_pandas(categories=None, ordered=False):
+    return pd.api.types.CategoricalDtype(categories=categories, ordered=ordered)
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 706c8bd20..9bb08d69e 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -58,6 +58,7 @@ from .utils import (
     insert_meta_param_description,
     raise_on_meta_error,
     clear_known_categories,
+    group_split_dispatch,
     is_categorical_dtype,
     has_known_categories,
     PANDAS_VERSION,
@@ -498,7 +499,9 @@ Dask Name: {name}, {task} tasks"""
             raise ValueError(msg)
 
     @derived_from(pd.DataFrame)
-    def drop_duplicates(self, subset=None, split_every=None, split_out=1, **kwargs):
+    def drop_duplicates(
+        self, subset=None, split_every=None, split_out=1, ignore_index=False, **kwargs
+    ):
         if subset is not None:
             # Let pandas error on bad inputs
             self._meta_nonempty.drop_duplicates(subset=subset, **kwargs)
@@ -523,6 +526,7 @@ Dask Name: {name}, {task} tasks"""
             split_out=split_out,
             split_out_setup=split_out_setup,
             split_out_setup_kwargs=split_out_setup_kwargs,
+            ignore_index=ignore_index,
             **kwargs
         )
 
@@ -4716,7 +4720,9 @@ def _maybe_from_pandas(dfs):
     return dfs
 
 
-def hash_shard(df, nparts, split_out_setup=None, split_out_setup_kwargs=None):
+def hash_shard(
+    df, nparts, split_out_setup=None, split_out_setup_kwargs=None, ignore_index=False
+):
     if split_out_setup:
         h = split_out_setup(df, **(split_out_setup_kwargs or {}))
     else:
@@ -4725,8 +4731,8 @@ def hash_shard(df, nparts, split_out_setup=None, split_out_setup_kwargs=None):
     h = hash_object_dispatch(h, index=False)
     if is_series_like(h):
         h = h.values
-    h %= nparts
-    return {i: df.iloc[h == i] for i in range(nparts)}
+    np.mod(h, nparts, out=h)
+    return group_split_dispatch(df, h, nparts, ignore_index=ignore_index)
 
 
 def split_evenly(df, k):
@@ -4762,6 +4768,7 @@ def apply_concat_apply(
     split_out_setup=None,
     split_out_setup_kwargs=None,
     sort=None,
+    ignore_index=False,
     **kwargs
 ):
     """Apply a function to blocks, then concat, then apply again
@@ -4803,6 +4810,8 @@ def apply_concat_apply(
         Keywords for the `split_out_setup` function only.
     sort : bool, default None
         If allowed, sort the keys of the output aggregation.
+    ignore_index : bool, default False
+        If True, do not preserve index values throughout ACA operations.
     kwargs :
         All remaining keywords will be passed to ``chunk``, ``aggregate``, and
         ``combine``.
@@ -4892,6 +4901,7 @@ def apply_concat_apply(
                 split_out,
                 split_out_setup,
                 split_out_setup_kwargs,
+                ignore_index,
             )
             for j in range(split_out):
                 dsk[(shard_prefix, 0, i, j)] = (getitem, (split_prefix, i), j)
@@ -4906,7 +4916,7 @@ def apply_concat_apply(
     while k > split_every:
         for part_i, inds in enumerate(partition_all(split_every, range(k))):
             for j in range(split_out):
-                conc = (_concat, [(a, depth, i, j) for i in inds])
+                conc = (_concat, [(a, depth, i, j) for i in inds], ignore_index)
                 if combine_kwargs:
                     dsk[(b, depth + 1, part_i, j)] = (
                         apply,
@@ -4932,7 +4942,7 @@ def apply_concat_apply(
     # Aggregate
     for j in range(split_out):
         b = "{0}-agg-{1}".format(token or funcname(aggregate), token_key)
-        conc = (_concat, [(a, depth, i, j) for i in range(k)])
+        conc = (_concat, [(a, depth, i, j) for i in range(k)], ignore_index)
         if aggregate_kwargs:
             dsk[(b, j)] = (apply, aggregate, [conc], aggregate_kwargs)
         else:
@@ -4940,7 +4950,9 @@ def apply_concat_apply(
 
     if meta is no_default:
         meta_chunk = _emulate(chunk, *args, udf=True, **chunk_kwargs)
-        meta = _emulate(aggregate, _concat([meta_chunk]), udf=True, **aggregate_kwargs)
+        meta = _emulate(
+            aggregate, _concat([meta_chunk], ignore_index), udf=True, **aggregate_kwargs
+        )
     meta = make_meta(
         meta, index=(getattr(make_meta(dfs[0]), "index", None) if dfs else None)
     )
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index b75a764d0..2fae52a1e 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -546,9 +546,34 @@ class ArrowEngine(Engine):
             use_pandas_metadata=True,
             use_threads=False,
             **kwargs.get("read", {}),
-        ).to_pandas(categories=categories, use_threads=False, ignore_metadata=True)[
-            list(columns_and_parts)
-        ]
+        ).to_pandas(categories=categories, use_threads=False, ignore_metadata=False)
+
+        # Note that `to_pandas(ignore_metadata=False)` means
+        # pyarrow will use the pandas metadata to set the index.
+        index_in_columns_and_parts = set(df.index.names).issubset(
+            set(columns_and_parts)
+        )
+        if not index:
+            if index_in_columns_and_parts:
+                # User does not want to set index and a desired
+                # column/partition has been set to the index
+                df.reset_index(drop=False, inplace=True)
+            else:
+                # User does not want to set index and an
+                # "unwanted" column has been set to the index
+                df.reset_index(drop=True, inplace=True)
+        else:
+            if set(df.index.names) != set(index) and index_in_columns_and_parts:
+                # The wrong index has been set and it contains
+                # one or more desired columns/partitions
+                df.reset_index(drop=False, inplace=True)
+            elif index_in_columns_and_parts:
+                # The correct index has already been set
+                index = False
+                columns_and_parts = list(
+                    set(columns_and_parts).difference(set(df.index.names))
+                )
+        df = df[list(columns_and_parts)]
 
         if index:
             df = df.set_index(index)
diff --git a/dask/dataframe/io/parquet/fastparquet.py b/dask/dataframe/io/parquet/fastparquet.py
index 6ef9534e9..7b97f8745 100644
--- a/dask/dataframe/io/parquet/fastparquet.py
+++ b/dask/dataframe/io/parquet/fastparquet.py
@@ -119,11 +119,14 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
         if gather_statistics is not False:
             # This scans all the files, allowing index/divisions
             # and filtering
-            pf = ParquetFile(
-                paths, open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
-            )
             if "_metadata" not in fns:
+                paths_use = paths
                 fast_metadata = False
+            else:
+                paths_use = base + fs.sep + "_metadata"
+            pf = ParquetFile(
+                paths_use, open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
+            )
         else:
             if "_metadata" in fns:
                 # We have a _metadata file, lets use it
@@ -177,11 +180,12 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
             parts = paths.copy()
     else:
         # There is only one file to read
+        base = None
         pf = ParquetFile(
             paths[0], open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
         )
 
-    return parts, pf, gather_statistics, fast_metadata
+    return parts, pf, gather_statistics, fast_metadata, base
 
 
 class FastParquetEngine(Engine):
@@ -199,7 +203,7 @@ class FastParquetEngine(Engine):
         # Also, initialize `parts`.  If `parts` is populated here,
         # then each part will correspond to a file.  Otherwise, each part will
         # correspond to a row group (populated below).
-        parts, pf, gather_statistics, fast_metadata = _determine_pf_parts(
+        parts, pf, gather_statistics, fast_metadata, base_path = _determine_pf_parts(
             fs, paths, gather_statistics, **kwargs
         )
 
@@ -284,7 +288,6 @@ class FastParquetEngine(Engine):
                 meta[catcol] = meta[catcol].cat.set_categories(pf.cats[catcol])
             elif meta.index.name == catcol:
                 meta.index = meta.index.set_categories(pf.cats[catcol])
-
         if gather_statistics and pf.row_groups:
             stats = []
             if filters is None:
@@ -341,28 +344,97 @@ class FastParquetEngine(Engine):
         pf._dtypes = lambda *args: pf.dtypes  # ugly patch, could be fixed
         pf.fmd.row_groups = None
 
+        # Constructing "piece" and "pf" for each dask partition. We will
+        # need to consider the following output scenarios:
+        #
+        #  1) Each "piece" is a file path, and "pf" is `None`
+        #      - `gather_statistics==False` and no "_metadata" available
+        #  2) Each "piece" is a row-group index, and "pf" is ParquetFile object
+        #      - We have parquet partitions and no "_metadata" available
+        #  3) Each "piece" is a row-group index, and "pf" is a `tuple`
+        #      - The value of the 0th tuple element depends on the following:
+        #      A) Dataset is partitioned and "_metadata" exists
+        #          - 0th tuple element will be the path to "_metadata"
+        #      B) Dataset is not partitioned
+        #          - 0th tuple element will be the path to the data
+        #      C) Dataset is partitioned and "_metadata" does not exist
+        #          - 0th tuple element will be the original `paths` argument
+        #            (We will let the workers use `_determine_pf_parts`)
+
         # Create `parts`
         # This is a list of row-group-descriptor dicts, or file-paths
         # if we have a list of files and gather_statistics=False
-        if not parts:
-            partsin = pf.row_groups
-            if fast_metadata:
-                pf = (paths, gather_statistics)
-        else:
+        base_path = (base_path or "") + fs.sep
+        if parts:
+            # Case (1)
+            # `parts` is just a list of path names.
+            pqpartitions = None
+            pf_deps = None
             partsin = parts
-            pf = None
+        else:
+            # Populate `partsin` with dataset row-groups
+            partsin = pf.row_groups
+            pqpartitions = pf.info.get("partitions", None)
+            if pqpartitions and not fast_metadata:
+                # Case (2)
+                # We have parquet partitions, and do not have
+                # a "_metadata" file for the worker to read.
+                # Therefore, we need to pass the pf object in
+                # the task graph
+                pf_deps = pf
+            else:
+                # Case (3)
+                # We don't need to pass a pf object in the task graph.
+                # Instead, we can try to pass the path for each part.
+                pf_deps = "tuple"
+
         parts = []
-        for i, piece in enumerate(partsin):
-            if pf and not fast_metadata:
-                for col in piece.columns:
+        i_path = 0
+        path_last = None
+
+        # Loop over DataFrame partitions.
+        # Each `part` will be a row-group or path ()
+        for i, part in enumerate(partsin):
+            if pqpartitions and fast_metadata:
+                # Case (3A)
+                # We can pass a "_metadata" path
+                file_path = base_path + "_metadata"
+                i_path = i
+            elif (
+                pf_deps
+                and isinstance(part.columns[0].file_path, str)
+                and not pqpartitions
+            ):
+                # Case (3B)
+                # We can pass a specific file/part path
+                path_curr = part.columns[0].file_path
+                if path_last and path_curr == path_last:
+                    i_path += 1
+                else:
+                    i_path = 0
+                    path_last = path_curr
+                file_path = base_path + path_curr
+            else:
+                # Case (3C)
+                # We cannot pass a specific file/part path
+                file_path = paths
+                i_path = i
+
+            # Strip down pf object
+            if pf_deps and pf_deps != "tuple":
+                # Case (2)
+                for col in part.columns:
                     col.meta_data.statistics = None
                     col.meta_data.encoding_stats = None
-            piece_item = i if pf else piece
-            part = {
-                "piece": piece_item,
-                "kwargs": {"pf": pf, "categories": categories_dict or categories},
+
+            # Final definition of "piece" and "pf" for this output partition
+            piece = i_path if pf_deps else part
+            pf_piece = (file_path, gather_statistics) if pf_deps == "tuple" else pf_deps
+            part_item = {
+                "piece": piece,
+                "kwargs": {"pf": pf_piece, "categories": categories_dict or categories},
             }
-            parts.append(part)
+            parts.append(part_item)
 
         return (meta, stats, parts)
 
@@ -385,7 +457,12 @@ class FastParquetEngine(Engine):
             return pf.to_pandas(columns, categories, index=index)
         else:
             if isinstance(pf, tuple):
-                pf = _determine_pf_parts(fs, pf[0], pf[1], **kwargs)[1]
+                if isinstance(pf[0], list):
+                    pf = _determine_pf_parts(fs, pf[0], pf[1], **kwargs)[1]
+                else:
+                    pf = ParquetFile(
+                        pf[0], open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
+                    )
                 pf._dtypes = lambda *args: pf.dtypes  # ugly patch, could be fixed
                 pf.fmd.row_groups = None
             rg_piece = pf.row_groups[piece]
diff --git a/dask/dataframe/shuffle.py b/dask/dataframe/shuffle.py
index 4553aa227..c2540b557 100644
--- a/dask/dataframe/shuffle.py
+++ b/dask/dataframe/shuffle.py
@@ -674,11 +674,9 @@ def set_partitions_pre(s, divisions):
 def shuffle_group_2(df, col, ignore_index):
     if not len(df):
         return {}, df
-    ind = df[col].astype(np.int64)
+    ind = df[col].astype(np.int32)
     n = ind.max() + 1
-    result2 = group_split_dispatch(
-        df, ind.values.view(np.int64), n, ignore_index=ignore_index
-    )
+    result2 = group_split_dispatch(df, ind.values.view(), n, ignore_index=ignore_index)
     return result2, df.iloc[:0]
 
 
@@ -728,7 +726,7 @@ def shuffle_group(df, col, stage, k, npartitions, ignore_index):
     np.floor_divide(c, k ** stage, out=c)
     np.mod(c, k, out=c)
 
-    return group_split_dispatch(df, c.astype(np.int64), k, ignore_index=ignore_index)
+    return group_split_dispatch(df, c, k, ignore_index=ignore_index)
 
 
 @contextlib.contextmanager
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index ae12096dc..6b07f5a80 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -478,7 +478,9 @@ group_split_dispatch = Dispatch("group_split_dispatch")
 
 @group_split_dispatch.register((pd.DataFrame, pd.Series, pd.Index))
 def group_split_pandas(df, c, k, ignore_index=False):
-    indexer, locations = pd._libs.algos.groupsort_indexer(c, k)
+    indexer, locations = pd._libs.algos.groupsort_indexer(
+        c.astype(np.int64, copy=False), k
+    )
     df2 = df.take(indexer)
     locations = locations.cumsum()
     parts = [df2.iloc[a:b] for a, b in zip(locations[:-1], locations[1:])]
diff --git a/dask/optimization.py b/dask/optimization.py
index eafc9eb87..3f8309c9d 100644
--- a/dask/optimization.py
+++ b/dask/optimization.py
@@ -447,7 +447,7 @@ def fuse(
     dependencies: dict, optional
         {key: [list-of-keys]}.  Must be a list to provide count of each key
         This optional input often comes from ``cull``
-    ave_width: float (default 2)
+    ave_width: float (default 1)
         Upper limit for ``width = num_nodes / height``, a good measure of
         parallelizability
     max_width: int
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 5f984712e..607d2a5c8 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -25,6 +25,11 @@ import os
 
 # Add any Sphinx extension module names here, as strings. They can be extensions
 # coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
+import sys
+
+source_dir = os.path.dirname(__file__)
+sys.path.insert(0, os.path.join(source_dir, "ext"))
+
 extensions = [
     "sphinx.ext.autodoc",
     "sphinx.ext.mathjax",
@@ -33,6 +38,7 @@ extensions = [
     "sphinx.ext.extlinks",
     "numpydoc",
     "sphinx_click.ext",
+    "dask_config_sphinx_ext",
 ]
 
 numpydoc_show_class_members = False
diff --git a/docs/source/configuration-reference.rst b/docs/source/configuration-reference.rst
new file mode 100644
index 000000000..38f66088d
--- /dev/null
+++ b/docs/source/configuration-reference.rst
@@ -0,0 +1,96 @@
+Configuration Reference
+=======================
+
+.. contents:: :local:
+
+Dask
+----
+
+.. dask-config-block::
+    :location: dask
+    :config: https://raw.githubusercontent.com/dask/dask/master/dask/dask.yaml
+    :schema: https://raw.githubusercontent.com/dask/dask/master/dask/dask-schema.yaml
+
+
+Distributed
+-----------
+
+Client
+~~~~~~
+
+.. dask-config-block::
+    :location: distributed.client
+    :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+    :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
+
+Comm
+~~~~
+
+.. dask-config-block::
+    :location: distributed.comm
+    :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+    :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
+
+
+Dashboard
+~~~~~~~~~
+
+.. dask-config-block::
+    :location: distributed.dashboard
+    :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+    :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
+
+
+Deploy
+~~~~~~
+
+.. dask-config-block::
+    :location: distributed.deploy
+    :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+    :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
+
+
+
+Scheduler
+~~~~~~~~~
+
+.. dask-config-block::
+    :location: distributed.scheduler
+    :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+    :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
+
+
+Worker
+~~~~~~
+
+.. dask-config-block::
+    :location: distributed.worker
+    :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+    :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
+
+
+Admin
+~~~~~
+
+.. dask-config-block::
+    :location: distributed.admin
+    :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+    :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
+
+
+UCX
+~~~
+
+.. dask-config-block::
+   :location: ucx
+   :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+   :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
+
+
+RMM
+~~~
+
+.. dask-config-block::
+    :location: rmm
+    :config: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed.yaml
+    :schema: https://raw.githubusercontent.com/dask/distributed/master/distributed/distributed-schema.yaml
diff --git a/docs/source/ext/dask_config_sphinx_ext.py b/docs/source/ext/dask_config_sphinx_ext.py
new file mode 100644
index 000000000..6a5bd3f85
--- /dev/null
+++ b/docs/source/ext/dask_config_sphinx_ext.py
@@ -0,0 +1,94 @@
+import requests
+import yaml
+
+from docutils import nodes
+from docutils.parsers.rst import Directive, directives
+
+
+def get_remote_yaml(url):
+    r = requests.get(url)
+    return yaml.safe_load(r.text)
+
+
+class DaskConfigDirective(Directive):
+
+    option_spec = {
+        "location": directives.unchanged,
+        "schema": directives.uri,
+        "config": directives.uri,
+    }
+
+    def run(self):
+        location = self.options["location"]
+        config = self.options["config"]
+        schema = self.options["schema"]
+
+        config = get_remote_yaml(config)
+        schema = get_remote_yaml(schema)
+
+        for k in location.split("."):
+            # dask config does not have a top level key
+            # we need to pass full schema and config
+            if k == "dask":
+                schema = schema
+                config = config
+            else:
+                config = config[k]
+                schema = schema["properties"].get(k, {})
+        html = generate_html(config, schema, location)
+        return [nodes.raw("", html, format="html")]
+
+
+def setup(app):
+    app.add_directive("dask-config-block", DaskConfigDirective)
+
+    return {
+        "version": "0.1",
+        "parallel_read_safe": True,
+        "parallel_write_safe": True,
+    }
+
+
+def dask_config_to_html(key, value, schema, prefix=""):
+    if isinstance(value, dict):
+        return sum(
+            [
+                dask_config_to_html(
+                    k,
+                    v,
+                    schema.get("properties", {}).get(k, {"properties": {}}),
+                    prefix=prefix + key + ".",
+                )
+                for k, v in value.items()
+            ],
+            [],
+        )
+
+    else:
+
+        try:
+            description = schema["description"]
+            description = description.strip()
+        except KeyError:
+            description = "No Comment"
+
+        key = prefix + key
+        value = str(value)
+        node = f"""<dl class="py data">
+                     <dt id="{key}">
+                       <code class="sig-prename descclassname">{key}</code>
+                       <em class="property">&nbsp;&nbsp;{value}</p></em>
+                       <a class="headerlink" href="#{key}" title="Permalink to this definition">¶</a>
+                     </dt>
+                    <dd><p>{description}</p></dd>
+                    </dl>
+
+                """
+        return [node]
+
+
+def generate_html(config, schema, location):
+    nested_html = dask_config_to_html(
+        key="", value=config, schema=schema, prefix=location
+    )
+    return "".join(nested_html)
diff --git a/docs/source/index.rst b/docs/source/index.rst
index 24096fe2d..a8db5d8c0 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -179,6 +179,7 @@ messy situations in everyday problems.
    develop.rst
    changelog.rst
    configuration.rst
+   configuration-reference.rst
    educational-resources.rst
    presentations.rst
    cheatsheet.rst
diff --git a/docs/source/setup/kubernetes-helm.rst b/docs/source/setup/kubernetes-helm.rst
index 2576d3ed2..4691a85d9 100644
--- a/docs/source/setup/kubernetes-helm.rst
+++ b/docs/source/setup/kubernetes-helm.rst
@@ -47,7 +47,7 @@ You will need to add this to your known channels and update your local charts::
 
 Now, you can launch Dask on your Kubernetes cluster using the Dask Helm_ chart::
 
-   helm install dask/dask
+   helm install my-dask dask/dask
 
 This deploys a ``dask-scheduler``, several ``dask-worker`` processes, and
 also an optional Jupyter server.


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index 31513056e..2fe46b122 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -2377,6 +2377,20 @@ def test_from_array_dask_array():
         da.from_array(dx)
 
 
+def test_from_array_dask_collection_warns():
+    class CustomCollection(np.ndarray):
+        def __dask_graph__(self):
+            return {"bar": 1}
+
+    x = CustomCollection([1, 2, 3])
+    with pytest.warns(UserWarning):
+        da.from_array(x)
+
+    # Ensure da.array warns too
+    with pytest.warns(UserWarning):
+        da.array(x)
+
+
 def test_asarray():
     assert_eq(da.asarray([1, 2, 3]), np.asarray([1, 2, 3]))
 
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index cc4e6b1dd..1973209fa 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -2365,3 +2365,27 @@ def test_filter_nonpartition_columns(
     df_read = ddf_read.compute()
     assert len(df_read) == len(df_read[df_read["time"] < 5])
     assert df_read["time"].max() < 5
+
+
+def test_pandas_metadata_nullable_pyarrow(tmpdir):
+
+    check_pyarrow()
+    if pa.__version__ < LooseVersion("0.16.0") or pd.__version__ < LooseVersion(
+        "1.0.0"
+    ):
+        pytest.skip("PyArrow>=0.16 and Pandas>=1.0.0 Required.")
+    tmpdir = str(tmpdir)
+
+    ddf1 = dd.from_pandas(
+        pd.DataFrame(
+            {
+                "A": pd.array([1, None, 2], dtype="Int64"),
+                "B": pd.array(["dog", "cat", None], dtype="str"),
+            }
+        ),
+        npartitions=1,
+    )
+    ddf1.to_parquet(tmpdir, engine="pyarrow")
+    ddf2 = dd.read_parquet(tmpdir, engine="pyarrow")
+
+    assert_eq(ddf1, ddf2, check_index=False)
diff --git a/dask/dataframe/tests/test_categorical.py b/dask/dataframe/tests/test_categorical.py
index 07553c9e1..69f831a86 100644
--- a/dask/dataframe/tests/test_categorical.py
+++ b/dask/dataframe/tests/test_categorical.py
@@ -212,6 +212,24 @@ def test_categorize():
         ddf.categorize(split_every="foo")
 
 
+def test_categorical_dtype():
+    cat_dtype = dd.categorical.categorical_dtype(
+        meta=a, categories=["a", "b", "c"], ordered=False
+    )
+    assert_eq(cat_dtype.categories, pd.Index(["a", "b", "c"]))
+    assert_eq(cat_dtype.ordered, False)
+
+    cat_dtype = dd.categorical.categorical_dtype(meta=a, categories=["a", "b", "c"])
+    assert_eq(cat_dtype.categories, pd.Index(["a", "b", "c"]))
+    assert_eq(cat_dtype.ordered, False)
+
+    cat_dtype = dd.categorical.categorical_dtype(
+        meta=a, categories=[1, 100, 200], ordered=True
+    )
+    assert_eq(cat_dtype.categories, pd.Index([1, 100, 200]))
+    assert_eq(cat_dtype.ordered, True)
+
+
 def test_categorize_index():
     # Object dtype
     ddf = dd.from_pandas(_compat.makeDataFrame(), npartitions=5)
diff --git a/dask/tests/test_config.py b/dask/tests/test_config.py
index 480e0df78..addbd58bf 100644
--- a/dask/tests/test_config.py
+++ b/dask/tests/test_config.py
@@ -420,6 +420,52 @@ def test_core_file():
     assert "shuffle-compression" in dask.config.get("dataframe")
 
 
+def test_schema():
+    jsonschema = pytest.importorskip("jsonschema")
+
+    config_fn = os.path.join(os.path.dirname(__file__), "..", "dask.yaml")
+    schema_fn = os.path.join(os.path.dirname(__file__), "..", "dask-schema.yaml")
+
+    with open(config_fn) as f:
+        config = yaml.safe_load(f)
+
+    with open(schema_fn) as f:
+        schema = yaml.safe_load(f)
+
+    jsonschema.validate(config, schema)
+
+
+def test_schema_is_complete():
+    config_fn = os.path.join(os.path.dirname(__file__), "..", "dask.yaml")
+    schema_fn = os.path.join(os.path.dirname(__file__), "..", "dask-schema.yaml")
+
+    with open(config_fn) as f:
+        config = yaml.safe_load(f)
+
+    with open(schema_fn) as f:
+        schema = yaml.safe_load(f)
+
+    def test_matches(c, s):
+        for k, v in c.items():
+            if list(c) != list(s["properties"]):
+                raise ValueError(
+                    "\nThe dask.yaml and dask-schema.yaml files are not in sync.\n"
+                    "This usually happens when we add a new configuration value,\n"
+                    "but don't add the schema of that value to the dask-schema.yaml file\n"
+                    "Please modify these files to include the missing values: \n\n"
+                    "    dask.yaml:        {}\n"
+                    "    dask-schema.yaml: {}\n\n"
+                    "Examples in these files should be a good start, \n"
+                    "even if you are not familiar with the jsonschema spec".format(
+                        sorted(c), sorted(s["properties"])
+                    )
+                )
+            if isinstance(v, dict):
+                test_matches(c[k], s["properties"][k])
+
+    test_matches(config, schema)
+
+
 def test_deprecations():
     with pytest.warns(Warning) as info:
         with dask.config.set(fuse_ave_width=123):