diff --git a/dask/array/core.py b/dask/array/core.py
index 9346b1c04..15a63bd8b 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -2159,7 +2159,9 @@ def atop(func, out_ind, *args, **kwargs):
     concatenate : bool, keyword only
         If true concatenate arrays along dummy indices, else provide lists
     adjust_chunks : dict
-        Dictionary mapping index to function to be applied to chunk sizes
+        Dictionary mapping index to information to adjust chunk sizes.  Can
+        either be a constant chunksize, a tuple of all chunksizes, or a
+        function that converts old chunksize to new chunksize
     new_axes : dict, keyword only
         New indexes and their dimension lengths
 
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 4428e0ba5..62297b290 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -1,16 +1,15 @@
 from __future__ import absolute_import, division, print_function
 
-from collections import Iterable, Iterator, defaultdict
-from functools import wraps, partial
 import itertools
 import math
-from operator import getitem
 import types
 import uuid
-from random import Random
-from warnings import warn
+import warnings
+from collections import Iterable, Iterator, defaultdict
 from distutils.version import LooseVersion
-
+from functools import wraps, partial
+from operator import getitem
+from random import Random
 
 from toolz import (merge, take, reduce, valmap, map, partition_all, filter,
                    remove, compose, curry, first, second, accumulate, peek)
@@ -1050,7 +1049,7 @@ class Bag(Base):
 
         return type(self)(merge(self.dask, dsk), e, 1)
 
-    def take(self, k, npartitions=1, compute=True):
+    def take(self, k, npartitions=1, compute=True, warn=True):
         """ Take the first k elements.
 
         Parameters
@@ -1064,6 +1063,9 @@ class Bag(Base):
             returned. Pass -1 to use all partitions.
         compute : bool, optional
             Whether to compute the result, default is True.
+        warn : bool, optional
+            Whether to warn if the number of elements returned is less than
+            requested, default is True.
 
         >>> b = from_sequence(range(10))
         >>> b.take(3)  # doctest: +SKIP
@@ -1087,9 +1089,9 @@ class Bag(Base):
                 dsk[(name_p, i)] = (list, (take, k, (self.name, i)))
 
             concat = (toolz.concat, ([(name_p, i) for i in range(npartitions)]))
-            dsk[(name, 0)] = (safe_take, k, concat)
+            dsk[(name, 0)] = (safe_take, k, concat, warn)
         else:
-            dsk = {(name, 0): (safe_take, k, (self.name, 0))}
+            dsk = {(name, 0): (safe_take, k, (self.name, 0), warn)}
 
         b = Bag(merge(self.dask, dsk), name, 1)
 
@@ -1215,12 +1217,16 @@ class Bag(Base):
         import dask.dataframe as dd
         if meta is None:
             if isinstance(columns, pd.DataFrame):
-                warn("Passing metadata to `columns` is deprecated. Please "
-                     "use the `meta` keyword instead.")
+                warnings.warn("Passing metadata to `columns` is deprecated. "
+                              "Please use the `meta` keyword instead.")
                 meta = columns
             else:
-                head = self.take(1)[0]
-                meta = pd.DataFrame([head], columns=columns)
+                head = self.take(1, warn=False)
+                if len(head) == 0:
+                    raise ValueError("`dask.bag.Bag.to_dataframe` failed to "
+                                     "properly infer metadata, please pass in "
+                                     "metadata via the `meta` keyword")
+                meta = pd.DataFrame(list(head), columns=columns)
         elif columns is not None:
             raise ValueError("Can't specify both `meta` and `columns`")
         else:
@@ -1986,12 +1992,12 @@ def empty_safe_aggregate(func, parts, is_last):
     return empty_safe_apply(func, parts2, is_last)
 
 
-def safe_take(n, b):
+def safe_take(n, b, warn=True):
     r = list(take(n, b))
-    if len(r) != n:
-        warn("Insufficient elements for `take`. {0} elements requested, "
-             "only {1} elements available. Try passing larger `npartitions` "
-             "to `take`.".format(n, len(r)))
+    if len(r) != n and warn:
+        warnings.warn("Insufficient elements for `take`. {0} elements "
+                      "requested, only {1} elements available. Try passing "
+                      "larger `npartitions` to `take`.".format(n, len(r)))
     return r
 
 
diff --git a/docs/source/inspect.rst b/docs/source/inspect.rst
index 2cd98f57a..4725af80a 100644
--- a/docs/source/inspect.rst
+++ b/docs/source/inspect.rst
@@ -19,7 +19,7 @@ The first step is to look at the ``.dask`` attribute of an array
 
    >>> import dask.array as da
    >>> x = da.ones((5, 15), chunks=(5, 5))
-   >>> x.dask
+   >>> dict(x.dask)
    {('wrapped_1', 0, 0): (ones, (5, 5)),
     ('wrapped_1', 0, 1): (ones, (5, 5)),
     ('wrapped_1', 0, 2): (ones, (5, 5))}
@@ -29,7 +29,7 @@ objects
 
 .. code-block:: python
 
-   >>> (x + 1).dask
+   >>> dict((x + 1).dask)
    {('wrapped_1', 0, 0): (ones, (5, 5)),
     ('wrapped_1', 0, 1): (ones, (5, 5)),
     ('wrapped_1', 0, 2): (ones, (5, 5))
@@ -37,6 +37,9 @@ objects
     ('x_1', 0, 1): (add, ('wrapped_1', 0, 1), 1),
     ('x_1', 0, 2): (add, ('wrapped_1', 0, 2), 1)}
 
+.. note:: In this example we use simple names like ``x_1``, ``ones``, and
+  ``add`` for demonstration purposes.  However in practice these names may be
+  more complex and include long hashed names.
 
 Visualize graphs with DOT
 -------------------------

Dask objects inspection failing, issues with graph
Hi,

- I cannot reproduce the example given [here](http://dask.pydata.org/en/latest/inspect.html) on the dask website in order to inspect dask objects.
Please see the following notebook:
[inspect dask object](https://github.com/apatlpo/lops-array/blob/master/sandbox/test_graph.ipynb)

(sorry if github is not rendering notebooks as it the case for me right now)

- Also when I try to print graphs, complicated numbers show up which complicate the interpretation. For example:
[graph](https://github.com/apatlpo/lops-array/blob/master/sandbox/natl60_tseries_debug.ipynb)

Any thoughts?
Are these relates issues?


diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index 46427f714..44b874d48 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -530,17 +530,23 @@ def test_take_npartitions():
         b.take(1, npartitions=5)
 
 
-@pytest.mark.skipif(sys.version_info[:2] == (3,3),
-                    reason="Python3.3 uses pytest2.7.2, w/o warns method")
 def test_take_npartitions_warn():
-    with pytest.warns(None):
-        b.take(100)
+    # Use single-threaded scheduler so warnings are properly captured in the
+    # same process
+    with dask.set_options(get=dask.get):
+        with pytest.warns(UserWarning):
+            b.take(100)
+
+        with pytest.warns(UserWarning):
+            b.take(7)
 
-    with pytest.warns(None):
-        b.take(7)
+        with pytest.warns(None) as rec:
+            b.take(7, npartitions=2)
+        assert len(rec) == 0
 
-    with pytest.warns(None):
-        b.take(7, npartitions=2)
+        with pytest.warns(None) as rec:
+            b.take(7, warn=False)
+        assert len(rec) == 0
 
 
 def test_map_is_lazy():
@@ -775,6 +781,11 @@ def test_to_dataframe():
     with pytest.raises(ValueError):
         b.to_dataframe(columns=['a', 'b'], meta=sol)
 
+    # Inference fails if empty first partition
+    b2 = b.filter(lambda x: x['a'] > 200)
+    with pytest.raises(ValueError):
+        b2.to_dataframe()
+
     # Single column
     b = b.pluck('a')
     sol = sol[['a']]

pip install -e .[complete]
parse_log_pytest
requirements.txt
pytest
apt-get update apt-get install -y gcc
3.6
requirements/base.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs==22.2.0
certifi==2021.5.30
click==8.0.4
cloudpickle==2.2.1
-e git+https://github.com/dask/dask.git@9a1f92801eebeefba4481fa5c9fc00badeee148a#egg=dask
distributed==1.20.2
HeapDict==1.0.1
importlib-metadata==4.8.3
iniconfig==1.1.1
locket==1.0.0
msgpack-python==0.5.6
numpy==1.19.5
packaging==21.3
pandas==1.1.5
partd==1.2.0
pluggy==1.0.0
psutil==7.0.0
py==1.11.0
pyparsing==3.1.4
pytest==7.0.1
python-dateutil==2.9.0.post0
pytz==2025.2
six==1.17.0
sortedcontainers==2.4.0
tblib==1.7.0
tomli==1.2.3
toolz==0.12.0
tornado==6.1
typing_extensions==4.1.1
zict==2.1.0
zipp==3.6.0

name: dask
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - attrs==22.2.0
      - click==8.0.4
      - cloudpickle==2.2.1
      - distributed==1.20.2
      - heapdict==1.0.1
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - locket==1.0.0
      - msgpack-python==0.5.6
      - numpy==1.19.5
      - packaging==21.3
      - pandas==1.1.5
      - partd==1.2.0
      - pluggy==1.0.0
      - psutil==7.0.0
      - py==1.11.0
      - pyparsing==3.1.4
      - pytest==7.0.1
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - six==1.17.0
      - sortedcontainers==2.4.0
      - tblib==1.7.0
      - tomli==1.2.3
      - toolz==0.12.0
      - tornado==6.1
      - typing-extensions==4.1.1
      - zict==2.1.0
      - zipp==3.6.0
prefix: /opt/conda/envs/dask

dask/bag/tests/test_bag.py::test_take_npartitions_warn dask/bag/tests/test_bag.py::test_to_dataframe
dask/bag/tests/test_bag.py::test_Bag dask/bag/tests/test_bag.py::test_keys dask/bag/tests/test_bag.py::test_bag_map dask/bag/tests/test_bag.py::test_map_method dask/bag/tests/test_bag.py::test_starmap dask/bag/tests/test_bag.py::test_filter dask/bag/tests/test_bag.py::test_remove dask/bag/tests/test_bag.py::test_iter dask/bag/tests/test_bag.py::test_repr[str] dask/bag/tests/test_bag.py::test_repr[repr] dask/bag/tests/test_bag.py::test_pluck dask/bag/tests/test_bag.py::test_pluck_with_default dask/bag/tests/test_bag.py::test_unzip dask/bag/tests/test_bag.py::test_fold dask/bag/tests/test_bag.py::test_distinct dask/bag/tests/test_bag.py::test_frequencies dask/bag/tests/test_bag.py::test_topk dask/bag/tests/test_bag.py::test_topk_with_non_callable_key[1] dask/bag/tests/test_bag.py::test_topk_with_non_callable_key[2] dask/bag/tests/test_bag.py::test_topk_with_multiarg_lambda dask/bag/tests/test_bag.py::test_lambdas dask/bag/tests/test_bag.py::test_reductions dask/bag/tests/test_bag.py::test_reduction_names dask/bag/tests/test_bag.py::test_tree_reductions dask/bag/tests/test_bag.py::test_aggregation[1] dask/bag/tests/test_bag.py::test_aggregation[3] dask/bag/tests/test_bag.py::test_aggregation[4] dask/bag/tests/test_bag.py::test_non_splittable_reductions[1] dask/bag/tests/test_bag.py::test_non_splittable_reductions[10] dask/bag/tests/test_bag.py::test_std dask/bag/tests/test_bag.py::test_var dask/bag/tests/test_bag.py::test_join dask/bag/tests/test_bag.py::test_foldby dask/bag/tests/test_bag.py::test_foldby_tree_reduction dask/bag/tests/test_bag.py::test_map_partitions dask/bag/tests/test_bag.py::test_map_partitions_args_kwargs dask/bag/tests/test_bag.py::test_random_sample_size dask/bag/tests/test_bag.py::test_random_sample_prob_range dask/bag/tests/test_bag.py::test_random_sample_repeated_computation dask/bag/tests/test_bag.py::test_random_sample_different_definitions dask/bag/tests/test_bag.py::test_random_sample_random_state dask/bag/tests/test_bag.py::test_lazify_task dask/bag/tests/test_bag.py::test_lazify dask/bag/tests/test_bag.py::test_inline_singleton_lists dask/bag/tests/test_bag.py::test_take dask/bag/tests/test_bag.py::test_take_npartitions dask/bag/tests/test_bag.py::test_map_is_lazy dask/bag/tests/test_bag.py::test_can_use_dict_to_make_concrete dask/bag/tests/test_bag.py::test_read_text dask/bag/tests/test_bag.py::test_read_text_large dask/bag/tests/test_bag.py::test_read_text_encoding dask/bag/tests/test_bag.py::test_read_text_large_gzip dask/bag/tests/test_bag.py::test_from_sequence dask/bag/tests/test_bag.py::test_from_long_sequence dask/bag/tests/test_bag.py::test_product dask/bag/tests/test_bag.py::test_partition_collect dask/bag/tests/test_bag.py::test_groupby dask/bag/tests/test_bag.py::test_groupby_with_indexer dask/bag/tests/test_bag.py::test_groupby_with_npartitions_changed dask/bag/tests/test_bag.py::test_concat dask/bag/tests/test_bag.py::test_flatten dask/bag/tests/test_bag.py::test_concat_after_map dask/bag/tests/test_bag.py::test_args dask/bag/tests/test_bag.py::test_to_textfiles[gz-GzipFile] dask/bag/tests/test_bag.py::test_to_textfiles[-open] dask/bag/tests/test_bag.py::test_to_textfiles[bz2-BZ2File] dask/bag/tests/test_bag.py::test_to_textfiles_name_function_preserves_order dask/bag/tests/test_bag.py::test_to_textfiles_name_function_warn dask/bag/tests/test_bag.py::test_to_textfiles_encoding dask/bag/tests/test_bag.py::test_to_textfiles_inputs dask/bag/tests/test_bag.py::test_to_textfiles_endlines dask/bag/tests/test_bag.py::test_string_namespace dask/bag/tests/test_bag.py::test_string_namespace_with_unicode dask/bag/tests/test_bag.py::test_str_empty_split dask/bag/tests/test_bag.py::test_map_with_iterator_function dask/bag/tests/test_bag.py::test_ensure_compute_output_is_concrete dask/bag/tests/test_bag.py::test_bag_class_extend dask/bag/tests/test_bag.py::test_gh715 dask/bag/tests/test_bag.py::test_bag_compute_forward_kwargs dask/bag/tests/test_bag.py::test_to_delayed dask/bag/tests/test_bag.py::test_to_delayed_optimizes dask/bag/tests/test_bag.py::test_from_delayed dask/bag/tests/test_bag.py::test_from_delayed_iterator dask/bag/tests/test_bag.py::test_range dask/bag/tests/test_bag.py::test_zip[1] dask/bag/tests/test_bag.py::test_zip[7] dask/bag/tests/test_bag.py::test_zip[10] dask/bag/tests/test_bag.py::test_zip[28] dask/bag/tests/test_bag.py::test_repartition[1-1] dask/bag/tests/test_bag.py::test_repartition[1-2] dask/bag/tests/test_bag.py::test_repartition[1-7] dask/bag/tests/test_bag.py::test_repartition[1-11] dask/bag/tests/test_bag.py::test_repartition[1-23] dask/bag/tests/test_bag.py::test_repartition[2-1] dask/bag/tests/test_bag.py::test_repartition[2-2] dask/bag/tests/test_bag.py::test_repartition[2-7] dask/bag/tests/test_bag.py::test_repartition[2-11] dask/bag/tests/test_bag.py::test_repartition[2-23] dask/bag/tests/test_bag.py::test_repartition[5-1] dask/bag/tests/test_bag.py::test_repartition[5-2] dask/bag/tests/test_bag.py::test_repartition[5-7] dask/bag/tests/test_bag.py::test_repartition[5-11] dask/bag/tests/test_bag.py::test_repartition[5-23] dask/bag/tests/test_bag.py::test_repartition[12-1] dask/bag/tests/test_bag.py::test_repartition[12-2] dask/bag/tests/test_bag.py::test_repartition[12-7] dask/bag/tests/test_bag.py::test_repartition[12-11] dask/bag/tests/test_bag.py::test_repartition[12-23] dask/bag/tests/test_bag.py::test_repartition[23-1] dask/bag/tests/test_bag.py::test_repartition[23-2] dask/bag/tests/test_bag.py::test_repartition[23-7] dask/bag/tests/test_bag.py::test_repartition[23-11] dask/bag/tests/test_bag.py::test_repartition[23-23] dask/bag/tests/test_bag.py::test_repartition_names dask/bag/tests/test_bag.py::test_accumulate dask/bag/tests/test_bag.py::test_groupby_tasks dask/bag/tests/test_bag.py::test_groupby_tasks_names dask/bag/tests/test_bag.py::test_groupby_tasks_2[1000-20-100] dask/bag/tests/test_bag.py::test_groupby_tasks_2[12345-234-1042] dask/bag/tests/test_bag.py::test_groupby_tasks_3 dask/bag/tests/test_bag.py::test_to_textfiles_empty_partitions dask/bag/tests/test_bag.py::test_reduction_empty dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[1] dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[2] dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[4] dask/bag/tests/test_bag.py::test_reduction_with_non_comparable_objects dask/bag/tests/test_bag.py::test_empty dask/bag/tests/test_bag.py::test_bag_picklable dask/bag/tests/test_bag.py::test_msgpack_unicode dask/bag/tests/test_bag.py::test_bag_with_single_callable dask/bag/tests/test_bag.py::test_optimize_fuse_keys dask/bag/tests/test_bag.py::test_reductions_are_lazy dask/bag/tests/test_bag.py::test_repeated_groupby dask/bag/tests/test_bag.py::test_temporary_directory dask/bag/tests/test_bag.py::test_empty_bag dask/bag/tests/test_bag.py::test_bag_paths
dask/array/core.py docs/source/inspect.rst dask/bag/core.py
dask/array/core.py docs/source/inspect.rst dask/bag/core.py