Repository: dask/dask

Issue:
Keep original filenames in dask.dataframe.read_csv
For the data I am reading, the path (directory name) is an important trait, and this would be useful to access (possibly as an additional column, ```path_as_column = True```) or at the very least in the collection of delayed objects
```
import dask.dataframe as dd
dd.read_csv('s3://bucket_name/*/*/*.csv', collection=True)
```

![image](https://user-images.githubusercontent.com/116120/31818992-d702cd1a-b59a-11e7-9636-dfca278ea320.png)

The collection version of the command comes closer

```
all_dfs = dd.read_csv('s3://bucket_name/*/*/*.csv', collection=True)
print('key:', all_dfs[0].key)
print('value:', all_dfs[0].compute())
```
but returns an internal code as the key and doesn't seem to have the path (s3://bucket_name/actual_folder/subfolder/fancyfile.csv) anywhere
```
key: pandas_read_text-35c2999796309c2c92e6438c0ebcbba4
value:    Unnamed: 0   T   N   M   count
0           0  T1  N0  M0  0.4454
1           5  T2  N0  M0  0.4076
2           6  T2  N0  M1  0.0666
3           1  T1  N0  M1  0.0612
4          10  T3  N0  M0  0.0054
```







Discussion:

Patch:
diff --git a/dask/bytes/core.py b/dask/bytes/core.py
index 3bf0cbc5a..5b64d63ed 100644
--- a/dask/bytes/core.py
+++ b/dask/bytes/core.py
@@ -19,7 +19,7 @@ from ..utils import import_required, is_integer
 
 
 def read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,
-               sample=True, compression=None, **kwargs):
+               sample=True, compression=None, include_path=False, **kwargs):
     """Given a path or paths, return delayed objects that read from those paths.
 
     The path may be a filename like ``'2015-01-01.csv'`` or a globstring
@@ -50,6 +50,9 @@ def read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,
     sample : bool or int
         Whether or not to return a header sample. If an integer is given it is
         used as sample size, otherwise the default sample size is 10kB.
+    include_path : bool
+        Whether or not to include the path with the bytes representing a particular file.
+        Default is False.
     **kwargs : dict
         Extra options that make sense to a particular storage connection, e.g.
         host, port, username, password, etc.
@@ -58,6 +61,7 @@ def read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,
     --------
     >>> sample, blocks = read_bytes('2015-*-*.csv', delimiter=b'\\n')  # doctest: +SKIP
     >>> sample, blocks = read_bytes('s3://bucket/2015-*-*.csv', delimiter=b'\\n')  # doctest: +SKIP
+    >>> sample, paths, blocks = read_bytes('2015-*-*.csv', include_path=True)  # doctest: +SKIP
 
     Returns
     -------
@@ -66,6 +70,10 @@ def read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,
     blocks : list of lists of ``dask.Delayed``
         Each list corresponds to a file, and each delayed object computes to a
         block of bytes from that file.
+    paths : list of strings, only included if include_path is True
+        List of same length as blocks, where each item is the path to the file
+        represented in the corresponding block.
+
     """
     fs, fs_token, paths = get_fs_token_paths(urlpath, mode='rb',
                                              storage_options=kwargs)
@@ -106,15 +114,17 @@ def read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,
         token = tokenize(fs_token, delimiter, path, fs.ukey(path),
                          compression, offset)
         keys = ['read-block-%s-%s' % (o, token) for o in offset]
-        out.append([delayed_read(OpenFile(fs, path, compression=compression),
-                                 o, l, delimiter, dask_key_name=key)
-                    for o, key, l in zip(offset, keys, length)])
+        values = [delayed_read(OpenFile(fs, path, compression=compression),
+                               o, l, delimiter, dask_key_name=key)
+                  for o, key, l in zip(offset, keys, length)]
+        out.append(values)
 
     if sample:
         with OpenFile(fs, paths[0], compression=compression) as f:
             nbytes = 10000 if sample is True else sample
             sample = read_block(f, 0, nbytes, delimiter)
-
+    if include_path:
+        return sample, out, paths
     return sample, out
 
 
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 095064900..eb57b06d2 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -2474,7 +2474,7 @@ class DataFrame(_Frame):
             if self.divisions != key.divisions:
                 from .multi import _maybe_align_partitions
                 self, key = _maybe_align_partitions([self, key])
-            dsk = {(name, i): (M._getitem_array, (self._name, i), (key._name, i))
+            dsk = {(name, i): (M.__getitem__, (self._name, i), (key._name, i))
                    for i in range(self.npartitions)}
             return new_dd_object(merge(self.dask, key.dask, dsk), name,
                                  self, self.divisions)
diff --git a/dask/dataframe/io/csv.py b/dask/dataframe/io/csv.py
index 20de3824f..0b788f0af 100644
--- a/dask/dataframe/io/csv.py
+++ b/dask/dataframe/io/csv.py
@@ -8,6 +8,7 @@ try:
 except ImportError:
     psutil = None
 
+import numpy as np
 import pandas as pd
 try:
     from pandas.api.types import CategoricalDtype
@@ -35,7 +36,7 @@ else:
 
 
 def pandas_read_text(reader, b, header, kwargs, dtypes=None, columns=None,
-                     write_header=True, enforce=False):
+                     write_header=True, enforce=False, path=None):
     """ Convert a block of bytes to a Pandas DataFrame
 
     Parameters
@@ -50,6 +51,8 @@ def pandas_read_text(reader, b, header, kwargs, dtypes=None, columns=None,
         A dictionary of keyword arguments to be passed to ``reader``
     dtypes : dict
         DTypes to assign to columns
+    path : tuple
+        A tuple containing path column name, path to file, and all paths.
 
     See Also
     --------
@@ -68,6 +71,11 @@ def pandas_read_text(reader, b, header, kwargs, dtypes=None, columns=None,
         raise ValueError("Columns do not match", df.columns, columns)
     elif columns:
         df.columns = columns
+    if path:
+        colname, path, paths = path
+        code = paths.index(path)
+        df = df.assign(**{
+            colname: pd.Categorical.from_codes(np.full(len(df), code), paths)})
     return df
 
 
@@ -157,7 +165,7 @@ def coerce_dtypes(df, dtypes):
 
 def text_blocks_to_pandas(reader, block_lists, header, head, kwargs,
                           collection=True, enforce=False,
-                          specified_dtypes=None):
+                          specified_dtypes=None, path=None):
     """ Convert blocks of bytes to a dask.dataframe or other high-level object
 
     This accepts a list of lists of values of bytes where each list corresponds
@@ -179,6 +187,8 @@ def text_blocks_to_pandas(reader, block_lists, header, head, kwargs,
     kwargs : dict
         Keyword arguments to pass down to ``reader``
     collection: boolean, optional (defaults to True)
+    path : tuple, optional
+        A tuple containing column name for path and list of all paths
 
     Returns
     -------
@@ -215,21 +225,32 @@ def text_blocks_to_pandas(reader, block_lists, header, head, kwargs,
     columns = list(head.columns)
     delayed_pandas_read_text = delayed(pandas_read_text, pure=True)
     dfs = []
-    for blocks in block_lists:
+    colname, paths = path or (None, None)
+
+    for i, blocks in enumerate(block_lists):
         if not blocks:
             continue
+        if path:
+            path_info = (colname, paths[i], paths)
+        else:
+            path_info = None
         df = delayed_pandas_read_text(reader, blocks[0], header, kwargs,
                                       dtypes, columns, write_header=False,
-                                      enforce=enforce)
+                                      enforce=enforce, path=path_info)
+
         dfs.append(df)
         rest_kwargs = kwargs.copy()
         rest_kwargs.pop('skiprows', None)
         for b in blocks[1:]:
             dfs.append(delayed_pandas_read_text(reader, b, header, rest_kwargs,
                                                 dtypes, columns,
-                                                enforce=enforce))
+                                                enforce=enforce, path=path_info))
 
     if collection:
+        if path:
+            head = head.assign(**{
+                colname: pd.Categorical.from_codes(np.zeros(len(head)), paths)
+            })
         if len(unknown_categoricals):
             head = clear_known_categories(head, cols=unknown_categoricals)
         return from_delayed(dfs, head)
@@ -257,12 +278,15 @@ else:
 def read_pandas(reader, urlpath, blocksize=AUTO_BLOCKSIZE, collection=True,
                 lineterminator=None, compression=None, sample=256000,
                 enforce=False, assume_missing=False, storage_options=None,
+                include_path_column=False,
                 **kwargs):
     reader_name = reader.__name__
     if lineterminator is not None and len(lineterminator) == 1:
         kwargs['lineterminator'] = lineterminator
     else:
         lineterminator = '\n'
+    if include_path_column and isinstance(include_path_column, bool):
+        include_path_column = 'path'
     if 'index' in kwargs or 'index_col' in kwargs:
         raise ValueError("Keyword 'index' not supported "
                          "dd.{0}(...).set_index('my-index') "
@@ -282,6 +306,10 @@ def read_pandas(reader, urlpath, blocksize=AUTO_BLOCKSIZE, collection=True,
     if isinstance(kwargs.get('header'), list):
         raise TypeError("List of header rows not supported for "
                         "dd.{0}".format(reader_name))
+    if isinstance(kwargs.get('converters'), dict) and include_path_column:
+        path_converter = kwargs.get('converters').get(include_path_column, None)
+    else:
+        path_converter = None
 
     if blocksize and compression not in seekable_files:
         warn("Warning %s compression does not support breaking apart files\n"
@@ -294,11 +322,21 @@ def read_pandas(reader, urlpath, blocksize=AUTO_BLOCKSIZE, collection=True,
                                   compression)
 
     b_lineterminator = lineterminator.encode()
-    b_sample, values = read_bytes(urlpath, delimiter=b_lineterminator,
-                                  blocksize=blocksize,
-                                  sample=sample,
-                                  compression=compression,
-                                  **(storage_options or {}))
+    b_out = read_bytes(urlpath, delimiter=b_lineterminator,
+                       blocksize=blocksize,
+                       sample=sample,
+                       compression=compression,
+                       include_path=include_path_column,
+                       **(storage_options or {}))
+
+    if include_path_column:
+        b_sample, values, paths = b_out
+        if path_converter:
+            paths = [path_converter(path) for path in paths]
+        path = (include_path_column, paths)
+    else:
+        b_sample, values = b_out
+        path = None
 
     if not isinstance(values[0], (tuple, list)):
         values = [values]
@@ -321,8 +359,13 @@ def read_pandas(reader, urlpath, blocksize=AUTO_BLOCKSIZE, collection=True,
 
     header = b'' if header is None else parts[skiprows] + b_lineterminator
 
-    # Use sample to infer dtypes
+    # Use sample to infer dtypes and check for presense of include_path_column
     head = reader(BytesIO(b_sample), **kwargs)
+    if include_path_column and (include_path_column in head.columns):
+        raise ValueError("Files already contain the column name: %s, so the "
+                         "path column cannot use this name. Please set "
+                         "`include_path_column` to a unique name."
+                         % include_path_column)
 
     specified_dtypes = kwargs.get('dtype', {})
     if specified_dtypes is None:
@@ -336,7 +379,8 @@ def read_pandas(reader, urlpath, blocksize=AUTO_BLOCKSIZE, collection=True,
 
     return text_blocks_to_pandas(reader, values, header, head, kwargs,
                                  collection=collection, enforce=enforce,
-                                 specified_dtypes=specified_dtypes)
+                                 specified_dtypes=specified_dtypes,
+                                 path=path)
 
 
 READ_DOC_TEMPLATE = """
@@ -384,6 +428,10 @@ assume_missing : bool, optional
 storage_options : dict, optional
     Extra options that make sense for a particular storage connection, e.g.
     host, port, username, password, etc.
+include_path_column : bool or str, optional
+    Whether or not to include the path to each particular file. If True a new
+    column is added to the dataframe called ``path``. If str, sets new column
+    name. Default is False.
 **kwargs
     Extra keyword arguments to forward to :func:`pandas.{reader}`.
 
@@ -415,6 +463,7 @@ def make_reader(reader, reader_name, file_type):
     def read(urlpath, blocksize=AUTO_BLOCKSIZE, collection=True,
              lineterminator=None, compression=None, sample=256000,
              enforce=False, assume_missing=False, storage_options=None,
+             include_path_column=False,
              **kwargs):
         return read_pandas(reader, urlpath, blocksize=blocksize,
                            collection=collection,
@@ -422,6 +471,7 @@ def make_reader(reader, reader_name, file_type):
                            compression=compression, sample=sample,
                            enforce=enforce, assume_missing=assume_missing,
                            storage_options=storage_options,
+                           include_path_column=include_path_column,
                            **kwargs)
     read.__doc__ = READ_DOC_TEMPLATE.format(reader=reader_name,
                                             file_type=file_type)
diff --git a/dask/dataframe/methods.py b/dask/dataframe/methods.py
index 3c4d863c9..67a48433d 100644
--- a/dask/dataframe/methods.py
+++ b/dask/dataframe/methods.py
@@ -222,7 +222,7 @@ else:
         return x._get_level_values(n)
 
 
-def concat(dfs, axis=0, join='outer', uniform=False):
+def concat(dfs, axis=0, join='outer', uniform=False, filter_warning=True):
     """Concatenate, handling some edge cases:
 
     - Unions categoricals between partitions
@@ -239,7 +239,10 @@ def concat(dfs, axis=0, join='outer', uniform=False):
         necessarily categories). Default is False.
     """
     if axis == 1:
-        return pd.concat(dfs, axis=axis, join=join)
+        with warnings.catch_warnings():
+            if filter_warning:
+                warnings.simplefilter('ignore', FutureWarning)
+            return pd.concat(dfs, axis=axis, join=join)
 
     if len(dfs) == 1:
         return dfs[0]
@@ -295,11 +298,14 @@ def concat(dfs, axis=0, join='outer', uniform=False):
             # pandas may raise a RuntimeWarning for comparing ints and strs
             with warnings.catch_warnings():
                 warnings.simplefilter("ignore", RuntimeWarning)
+                if filter_warning:
+                    warnings.simplefilter('ignore', FutureWarning)
                 cat_mask = pd.concat([(df.dtypes == 'category').to_frame().T
                                       for df in dfs3], join=join).any()
 
         if cat_mask.any():
             not_cat = cat_mask[~cat_mask].index
+            # this should be aligned, so no need to filter warning
             out = pd.concat([df[df.columns.intersection(not_cat)]
                              for df in dfs3], join=join)
             temp_ind = out.index
@@ -330,6 +336,8 @@ def concat(dfs, axis=0, join='outer', uniform=False):
             # pandas may raise a RuntimeWarning for comparing ints and strs
             with warnings.catch_warnings():
                 warnings.simplefilter("ignore", RuntimeWarning)
+                if filter_warning:
+                    warnings.simplefilter("ignore", FutureWarning)
                 out = pd.concat(dfs3, join=join)
     else:
         if is_categorical_dtype(dfs2[0].dtype):
@@ -337,7 +345,10 @@ def concat(dfs, axis=0, join='outer', uniform=False):
                 ind = concat([df.index for df in dfs2])
             return pd.Series(union_categoricals(dfs2), index=ind,
                              name=dfs2[0].name)
-        out = pd.concat(dfs2, join=join)
+        with warnings.catch_warnings():
+            if filter_warning:
+                warnings.simplefilter('ignore', FutureWarning)
+            out = pd.concat(dfs2, join=join)
     # Re-add the index if needed
     if ind is not None:
         out.index = ind
diff --git a/dask/dataframe/multi.py b/dask/dataframe/multi.py
index 51ca51eff..af3e2fee5 100644
--- a/dask/dataframe/multi.py
+++ b/dask/dataframe/multi.py
@@ -56,7 +56,7 @@ We proceed with hash joins in the following stages:
 from __future__ import absolute_import, division, print_function
 
 from functools import wraps, partial
-from warnings import warn
+import warnings
 
 from toolz import merge_sorted, unique, first
 import toolz
@@ -445,7 +445,9 @@ def concat_unindexed_dataframes(dfs):
 
 def concat_indexed_dataframes(dfs, axis=0, join='outer'):
     """ Concatenate indexed dataframes together along the index """
-    meta = methods.concat([df._meta for df in dfs], axis=axis, join=join)
+    warn = axis != 0
+    meta = methods.concat([df._meta for df in dfs], axis=axis, join=join,
+                          filter_warning=warn)
     empties = [strip_unknown_categories(df._meta) for df in dfs]
 
     dfs2, divisions, parts = align_partitions(*dfs)
@@ -456,7 +458,11 @@ def concat_indexed_dataframes(dfs, axis=0, join='outer'):
                for df, empty in zip(part, empties)]
               for part in parts]
 
-    dsk = dict(((name, i), (methods.concat, part, axis, join))
+    filter_warning = True
+    uniform = False
+
+    dsk = dict(((name, i), (methods.concat, part, axis, join,
+                            uniform, filter_warning))
                for i, part in enumerate(parts2))
     for df in dfs2:
         dsk.update(df.dask)
@@ -466,7 +472,7 @@ def concat_indexed_dataframes(dfs, axis=0, join='outer'):
 
 def stack_partitions(dfs, divisions, join='outer'):
     """Concatenate partitions on axis=0 by doing a simple stack"""
-    meta = methods.concat([df._meta for df in dfs], join=join)
+    meta = methods.concat([df._meta for df in dfs], join=join, filter_warning=False)
     empty = strip_unknown_categories(meta)
 
     name = 'concat-{0}'.format(tokenize(*dfs))
@@ -599,9 +605,10 @@ def concat(dfs, axis=0, join='outer', interleave_partitions=False):
         elif (len(dasks) == len(dfs) and
               all(not df.known_divisions for df in dfs) and
               len({df.npartitions for df in dasks}) == 1):
-            warn("Concatenating dataframes with unknown divisions.\n"
-                 "We're assuming that the indexes of each dataframes are \n"
-                 "aligned. This assumption is not generally safe.")
+            warnings.warn("Concatenating dataframes with unknown divisions.\n"
+                          "We're assuming that the indexes of each dataframes"
+                          " are \n aligned. This assumption is not generally "
+                          "safe.")
             return concat_unindexed_dataframes(dfs)
         else:
             raise ValueError('Unable to concatenate DataFrame with unknown '
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index c6014d50a..a7dc47b6f 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -477,7 +477,10 @@ def check_meta(x, meta, funcname=None, numeric_equal=True):
         errmsg = ("Expected partition of type `%s` but got "
                   "`%s`" % (type(meta).__name__, type(x).__name__))
     elif isinstance(meta, pd.DataFrame):
-        dtypes = pd.concat([x.dtypes, meta.dtypes], axis=1)
+        kwargs = dict()
+        if PANDAS_VERSION >= LooseVersion('0.23.0'):
+            kwargs['sort'] = True
+        dtypes = pd.concat([x.dtypes, meta.dtypes], axis=1, **kwargs)
         bad = [(col, a, b) for col, a, b in dtypes.fillna('-').itertuples()
                if not equal_dtypes(a, b)]
         if not bad:
diff --git a/dask/delayed.py b/dask/delayed.py
index 9fd315c91..f89e565c8 100644
--- a/dask/delayed.py
+++ b/dask/delayed.py
@@ -147,7 +147,7 @@ def delayed(obj, name=None, pure=None, nout=None, traverse=True):
         into ``nout`` objects, allowing for unpacking of results. By default
         iteration over ``Delayed`` objects will error. Note, that ``nout=1``
         expects ``obj``, to return a tuple of length 1, and consequently for
-        `nout=0``, ``obj`` should return an empty tuple.
+        ``nout=0``, ``obj`` should return an empty tuple.
     traverse : bool, optional
         By default dask traverses builtin python collections looking for dask
         objects passed to ``delayed``. For large collections this can be


Test patch:
diff --git a/dask/bytes/tests/test_local.py b/dask/bytes/tests/test_local.py
index b75914605..1fbdc3264 100644
--- a/dask/bytes/tests/test_local.py
+++ b/dask/bytes/tests/test_local.py
@@ -159,6 +159,12 @@ def test_read_bytes_blocksize_float():
             read_bytes('.test.account*', blocksize=5.5)
 
 
+def test_read_bytes_include_path():
+    with filetexts(files, mode='b'):
+        _, _, paths = read_bytes('.test.accounts.*', include_path=True)
+        assert {os.path.split(path)[1] for path in paths} == set(files.keys())
+
+
 def test_with_urls():
     with filetexts(files, mode='b'):
         # OS-independent file:// URI with glob *
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index 9948d0d88..80c0b6fd0 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -29,6 +29,10 @@ def normalize_text(s):
     return '\n'.join(map(str.strip, s.strip().split('\n')))
 
 
+def parse_filename(path):
+    return os.path.split(path)[1]
+
+
 csv_text = """
 name,amount
 Alice,100
@@ -284,6 +288,56 @@ def test_read_csv_files_list(dd_read, pd_read, files):
             dd_read([])
 
 
+@pytest.mark.parametrize('dd_read,files',
+                         [(dd.read_csv, csv_files),
+                          (dd.read_table, tsv_files)])
+def test_read_csv_include_path_column(dd_read, files):
+    with filetexts(files, mode='b'):
+        df = dd_read('2014-01-*.csv', include_path_column=True,
+                     converters={'path': parse_filename})
+        filenames = df.path.compute().unique()
+        assert '2014-01-01.csv' in filenames
+        assert '2014-01-02.csv' not in filenames
+        assert '2014-01-03.csv' in filenames
+
+
+@pytest.mark.parametrize('dd_read,files',
+                         [(dd.read_csv, csv_files),
+                          (dd.read_table, tsv_files)])
+def test_read_csv_include_path_column_as_str(dd_read, files):
+    with filetexts(files, mode='b'):
+        df = dd_read('2014-01-*.csv', include_path_column='filename',
+                     converters={'filename': parse_filename})
+        filenames = df.filename.compute().unique()
+        assert '2014-01-01.csv' in filenames
+        assert '2014-01-02.csv' not in filenames
+        assert '2014-01-03.csv' in filenames
+
+
+@pytest.mark.parametrize('dd_read,files',
+                         [(dd.read_csv, csv_files),
+                          (dd.read_table, tsv_files)])
+def test_read_csv_include_path_column_with_duplicate_name(dd_read, files):
+    with filetexts(files, mode='b'):
+        with pytest.raises(ValueError):
+            dd_read('2014-01-*.csv', include_path_column='name')
+
+
+@pytest.mark.parametrize('dd_read,files',
+                         [(dd.read_csv, csv_files),
+                          (dd.read_table, tsv_files)])
+def test_read_csv_include_path_column_is_dtype_category(dd_read, files):
+    with filetexts(files, mode='b'):
+        df = dd_read('2014-01-*.csv', include_path_column=True)
+        assert df.path.dtype == 'category'
+        assert has_known_categories(df.path)
+
+        dfs = dd_read('2014-01-*.csv', include_path_column=True, collection=False)
+        result = dfs[0].compute()
+        assert result.path.dtype == 'category'
+        assert has_known_categories(result.path)
+
+
 # After this point, we test just using read_csv, as all functionality
 # for both is implemented using the same code.
 
diff --git a/dask/dataframe/tests/test_multi.py b/dask/dataframe/tests/test_multi.py
index 370631e15..2262923b7 100644
--- a/dask/dataframe/tests/test_multi.py
+++ b/dask/dataframe/tests/test_multi.py
@@ -1,3 +1,5 @@
+import warnings
+
 import dask.dataframe as dd
 import numpy as np
 import pandas as pd
@@ -247,14 +249,25 @@ def test_indexed_concat(join):
                      index=[1, 2, 4, 5, 6, 8])
     b = dd.repartition(B, [1, 2, 5, 8])
 
-    result = concat_indexed_dataframes([a, b], join=join)
-    expected = pd.concat([A, B], axis=0, join=join)
+    with warnings.catch_warnings(record=True) as w:
+        expected = pd.concat([A, B], axis=0, join=join)
+
+    ctx = FutureWarning if w else None
+
+    with pytest.warns(ctx):
+        result = concat_indexed_dataframes([a, b], join=join)
     assert_eq(result, expected)
 
-    assert (sorted(concat_indexed_dataframes([a, b], join=join).dask) ==
-            sorted(concat_indexed_dataframes([a, b], join=join).dask))
-    assert (sorted(concat_indexed_dataframes([a, b], join='inner').dask) !=
-            sorted(concat_indexed_dataframes([a, b], join='outer').dask))
+    with warnings.catch_warnings():
+        warnings.simplefilter('ignore', FutureWarning)
+        assert (
+            sorted(concat_indexed_dataframes([a, b], join=join).dask) ==
+            sorted(concat_indexed_dataframes([a, b], join=join).dask)
+        )
+        assert (
+            sorted(concat_indexed_dataframes([a, b], join='inner').dask) !=
+            sorted(concat_indexed_dataframes([a, b], join='outer').dask)
+        )
 
 
 @pytest.mark.parametrize('join', ['inner', 'outer'])
@@ -276,8 +289,15 @@ def test_concat(join):
 
     for (dd1, dd2, pd1, pd2) in [(ddf1, ddf2, pdf1, pdf2),
                                  (ddf1, ddf3, pdf1, pdf3)]:
-        result = dd.concat([dd1, dd2], join=join)
-        expected = pd.concat([pd1, pd2], join=join)
+
+        with warnings.catch_warnings(record=True) as a:
+            expected = pd.concat([pd1, pd2], join=join)
+
+        # warn iff pandas warns
+        ctx = FutureWarning if a else None
+
+        with pytest.warns(ctx):
+            result = dd.concat([dd1, dd2], join=join)
         assert_eq(result, expected)
 
     # test outer only, inner has a problem on pandas side
@@ -287,8 +307,13 @@ def test_concat(join):
                                  (ddf1.x, ddf3.z, pdf1.x, pdf3.z),
                                  (ddf1.x, ddf2.x, pdf1.x, pdf2.x),
                                  (ddf1.x, ddf3.z, pdf1.x, pdf3.z)]:
-        result = dd.concat([dd1, dd2])
-        expected = pd.concat([pd1, pd2])
+        with warnings.catch_warnings(record=True) as a:
+            expected = pd.concat([pd1, pd2])
+
+        ctx = FutureWarning if a else None
+
+        with pytest.warns(ctx):
+            result = dd.concat([dd1, dd2])
         assert_eq(result, expected)
 
 
@@ -865,19 +890,38 @@ def test_concat2():
     cases = [[a, b], [a, c], [a, d]]
     assert dd.concat([a]) is a
     for case in cases:
-        result = dd.concat(case)
         pdcase = [_c.compute() for _c in case]
 
+        with warnings.catch_warnings(record=True) as w:
+            expected = pd.concat(pdcase)
+
+        ctx = FutureWarning if w else None
+
+        with pytest.warns(ctx):
+            result = dd.concat(case)
+
         assert result.npartitions == case[0].npartitions + case[1].npartitions
         assert result.divisions == (None, ) * (result.npartitions + 1)
-        assert_eq(pd.concat(pdcase), result)
-        assert set(result.dask) == set(dd.concat(case).dask)
+        assert_eq(expected, result)
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", FutureWarning)
+            assert set(result.dask) == set(dd.concat(case).dask)
+
+        with warnings.catch_warnings(record=True) as w:
+            expected = pd.concat(pdcase, join='inner')
+
+        ctx = FutureWarning if w else None
 
-        result = dd.concat(case, join='inner')
+        with pytest.warns(ctx):
+            result = dd.concat(case, join='inner')
         assert result.npartitions == case[0].npartitions + case[1].npartitions
         assert result.divisions == (None, ) * (result.npartitions + 1)
-        assert_eq(pd.concat(pdcase, join='inner'), result)
-        assert set(result.dask) == set(dd.concat(case, join='inner').dask)
+        assert_eq(result, result)
+
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore', FutureWarning)
+            assert set(result.dask) == set(dd.concat(case, join='inner').dask)
 
 
 def test_concat3():
@@ -891,25 +935,43 @@ def test_concat3():
     ddf2 = dd.from_pandas(pdf2, 3)
     ddf3 = dd.from_pandas(pdf3, 2)
 
-    result = dd.concat([ddf1, ddf2])
+    with warnings.catch_warnings(record=True) as w:
+        expected = pd.concat([pdf1, pdf2])
+
+    ctx = FutureWarning if w else None
+
+    with pytest.warns(ctx):
+        result = dd.concat([ddf1, ddf2])
+
     assert result.divisions == ddf1.divisions[:-1] + ddf2.divisions
     assert result.npartitions == ddf1.npartitions + ddf2.npartitions
-    assert_eq(result, pd.concat([pdf1, pdf2]))
+    assert_eq(result, expected)
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", FutureWarning)
+        assert_eq(dd.concat([ddf1, ddf2], interleave_partitions=True),
+                  pd.concat([pdf1, pdf2]))
+
+    with warnings.catch_warnings(record=True) as w:
+        expected = pd.concat([pdf1, pdf2, pdf3])
 
-    assert_eq(dd.concat([ddf1, ddf2], interleave_partitions=True),
-              pd.concat([pdf1, pdf2]))
+    ctx = FutureWarning if w else None
 
-    result = dd.concat([ddf1, ddf2, ddf3])
+    with pytest.warns(ctx):
+        result = dd.concat([ddf1, ddf2, ddf3])
     assert result.divisions == (ddf1.divisions[:-1] + ddf2.divisions[:-1] +
                                 ddf3.divisions)
     assert result.npartitions == (ddf1.npartitions + ddf2.npartitions +
                                   ddf3.npartitions)
-    assert_eq(result, pd.concat([pdf1, pdf2, pdf3]))
+    assert_eq(result, expected)
 
-    assert_eq(dd.concat([ddf1, ddf2, ddf3], interleave_partitions=True),
-              pd.concat([pdf1, pdf2, pdf3]))
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", FutureWarning)
+        assert_eq(dd.concat([ddf1, ddf2, ddf3], interleave_partitions=True),
+                  pd.concat([pdf1, pdf2, pdf3]))
 
 
+@pytest.mark.filterwarnings('ignore')
 def test_concat4_interleave_partitions():
     pdf1 = pd.DataFrame(np.random.randn(10, 5),
                         columns=list('ABCDE'), index=list('abcdefghij'))
@@ -946,6 +1008,7 @@ def test_concat4_interleave_partitions():
     assert msg in str(err.value)
 
 
+@pytest.mark.filterwarnings('ignore')
 def test_concat5():
     pdf1 = pd.DataFrame(np.random.randn(7, 5),
                         columns=list('ABCDE'), index=list('abcdefg'))
@@ -1010,6 +1073,7 @@ def test_concat5():
                          [(True, True, False), (True, False, True),
                           (True, False, False), (False, True, False),
                           (False, False, True), (False, False, False)])
+@pytest.mark.filterwarnings("ignore")
 def test_concat_categorical(known, cat_index, divisions):
     frames = [pd.DataFrame({'w': list('xxxxx'),
                             'x': np.arange(5),
@@ -1109,20 +1173,31 @@ def test_append():
     ddf3 = dd.from_pandas(df3, 2)
 
     s = pd.Series([7, 8], name=6, index=['a', 'b'])
-    assert_eq(ddf.append(s), df.append(s))
 
-    assert_eq(ddf.append(ddf2), df.append(df2))
-    assert_eq(ddf.a.append(ddf2.a), df.a.append(df2.a))
+    def check_with_warning(dask_obj, dask_append, pandas_obj, pandas_append):
+        with warnings.catch_warnings(record=True) as w:
+            expected = pandas_obj.append(pandas_append)
+
+        ctx = FutureWarning if w else None
+        with pytest.warns(ctx):
+            result = dask_obj.append(dask_append)
+
+        assert_eq(result, expected)
+
+    check_with_warning(ddf, s, df, s)
+    check_with_warning(ddf, ddf2, df, df2)
+    check_with_warning(ddf.a, ddf2.a, df.a, df2.a)
+
     # different columns
-    assert_eq(ddf.append(ddf3), df.append(df3))
-    assert_eq(ddf.a.append(ddf3.b), df.a.append(df3.b))
+    check_with_warning(ddf, ddf3, df, df3)
+    check_with_warning(ddf.a, ddf3.b, df.a, df3.b)
 
     # dask + pandas
-    assert_eq(ddf.append(df2), df.append(df2))
-    assert_eq(ddf.a.append(df2.a), df.a.append(df2.a))
+    check_with_warning(ddf, df2, df, df2)
+    check_with_warning(ddf.a, df2.a, df.a, df2.a)
 
-    assert_eq(ddf.append(df3), df.append(df3))
-    assert_eq(ddf.a.append(df3.b), df.a.append(df3.b))
+    check_with_warning(ddf, df3, df, df3)
+    check_with_warning(ddf.a, df3.b, df.a, df3.b)
 
     df4 = pd.DataFrame({'a': [1, 2, 3, 4, 5, 6],
                         'b': [1, 2, 3, 4, 5, 6]},
@@ -1132,6 +1207,7 @@ def test_append():
         ddf.append(ddf4)
 
 
+@pytest.mark.filterwarnings("ignore")
 def test_append2():
     dsk = {('x', 0): pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]}),
            ('x', 1): pd.DataFrame({'a': [4, 5, 6], 'b': [3, 2, 1]}),