Repository: dask/dask

Issue:
Dataframe does not support duplicate column names perfectly
I just realized that dask datafrmaes, in contrast to pandas, does not support duplicate column names (perfectly).

```
import dask.array as da
import dask.dataframe as dd
import numpy as np
arr = da.from_array(np.arange(10).reshape(5,2), chunks=(5,2))
dd.from_dask_array(arr, columns=['a', 'a'])
```
raises the following error message:

<details>

```
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
~/anaconda3/envs/dask-ml-dev/lib/python3.6/site-packages/IPython/core/formatters.py in __call__(self, obj)
    700                 type_pprinters=self.type_printers,
    701                 deferred_pprinters=self.deferred_printers)
--> 702             printer.pretty(obj)
    703             printer.flush()
    704             return stream.getvalue()

~/anaconda3/envs/dask-ml-dev/lib/python3.6/site-packages/IPython/lib/pretty.py in pretty(self, obj)
    398                         if cls is not object \
    399                                 and callable(cls.__dict__.get('__repr__')):
--> 400                             return _repr_pprint(obj, self, cycle)
    401 
    402             return _default_pprint(obj, self, cycle)

~/anaconda3/envs/dask-ml-dev/lib/python3.6/site-packages/IPython/lib/pretty.py in _repr_pprint(obj, p, cycle)
    693     """A pprint that just redirects to the normal repr function."""
    694     # Find newlines and replace them with p.break_()
--> 695     output = repr(obj)
    696     for idx,output_line in enumerate(output.splitlines()):
    697         if idx:

~/anaconda3/envs/dask-ml-dev/lib/python3.6/site-packages/dask/dataframe/core.py in __repr__(self)
    392 
    393     def __repr__(self):
--> 394         data = self._repr_data.to_string(max_rows=5, show_dimensions=False)
    395         return """Dask {klass} Structure:
    396 {data}

~/anaconda3/envs/dask-ml-dev/lib/python3.6/site-packages/dask/dataframe/core.py in __getattr__(self, key)
   2518             return new_dd_object(merge(self.dask, dsk), name,
   2519                                  meta, self.divisions)
-> 2520         raise AttributeError("'DataFrame' object has no attribute %r" % key)
   2521 
   2522     def __dir__(self):

AttributeError: 'DataFrame' object has no attribute '_repr_data'

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
~/anaconda3/envs/dask-ml-dev/lib/python3.6/site-packages/IPython/core/formatters.py in __call__(self, obj)
    343             method = get_real_method(obj, self.print_method)
    344             if method is not None:
--> 345                 return method()
    346             return None
    347         else:

~/anaconda3/envs/dask-ml-dev/lib/python3.6/site-packages/dask/dataframe/core.py in _repr_html_(self)
   3131 
   3132     def _repr_html_(self):
-> 3133         data = self._repr_data.to_html(max_rows=5,
   3134                                        show_dimensions=False, notebook=True)
   3135         return self._HTML_FMT.format(data=data, name=key_split(self._name),

~/anaconda3/envs/dask-ml-dev/lib/python3.6/site-packages/dask/dataframe/core.py in __getattr__(self, key)
   2518             return new_dd_object(merge(self.dask, dsk), name,
   2519                                  meta, self.divisions)
-> 2520         raise AttributeError("'DataFrame' object has no attribute %r" % key)
   2521 
   2522     def __dir__(self):

AttributeError: 'DataFrame' object has no attribute '_repr_data'
```

</details>

However, assigning the frame to a variable and computing the array works
```
m = dd.from_dask_array(arr, columns=['a', 'a'])
m.compute()

  | a | a
-- | -- | --
0 | 1
2 | 3
4 | 5
6 | 7
8 | 9

```

I didn't find the issue, so please apologies if it's a duplicate

Discussion:

Patch:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 510913a45..d478d9ff9 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -3124,8 +3124,8 @@ class DataFrame(_Frame):
     def _repr_data(self):
         meta = self._meta
         index = self._repr_divisions
-        values = {c: _repr_data_series(meta[c], index) for c in meta.columns}
-        return pd.DataFrame(values, columns=meta.columns)
+        series_list = [_repr_data_series(s, index=index) for _, s in meta.iteritems()]
+        return pd.concat(series_list, axis=1)
 
     _HTML_FMT = """<div><strong>Dask DataFrame Structure:</strong></div>
 {data}
diff --git a/dask/multiprocessing.py b/dask/multiprocessing.py
index aa825da6b..3a377b54c 100644
--- a/dask/multiprocessing.py
+++ b/dask/multiprocessing.py
@@ -160,6 +160,7 @@ def get(dsk, keys, num_workers=None, func_loads=None, func_dumps=None,
         If True [default], `fuse` is applied to the graph before computation.
     """
     pool = config.get('pool', None)
+    num_workers = num_workers or config.get('num_workers', None)
     if pool is None:
         context = get_context()
         pool = context.Pool(num_workers,
diff --git a/dask/threaded.py b/dask/threaded.py
index fd336db63..84a382e9c 100644
--- a/dask/threaded.py
+++ b/dask/threaded.py
@@ -56,6 +56,7 @@ def get(dsk, result, cache=None, num_workers=None, **kwargs):
     """
     global default_pool
     pool = config.get('pool', None)
+    num_workers = num_workers or config.get('num_workers', None)
     thread = current_thread()
 
     with pools_lock:


Test patch:
diff --git a/dask/dataframe/tests/test_format.py b/dask/dataframe/tests/test_format.py
index c9463eb25..3c9055737 100644
--- a/dask/dataframe/tests/test_format.py
+++ b/dask/dataframe/tests/test_format.py
@@ -3,6 +3,8 @@ import pandas as pd
 from textwrap import dedent
 
 import dask.dataframe as dd
+import dask.array as da
+import numpy as np
 from dask.dataframe.utils import PANDAS_VERSION
 
 if PANDAS_VERSION >= '0.21.0':
@@ -487,3 +489,9 @@ def test_categorical_format():
            "dtype: category\n"
            "Dask Name: from_pandas, 1 tasks")
     assert repr(unknown) == exp
+
+
+def test_duplicate_columns_repr():
+    arr = da.from_array(np.arange(10).reshape(5, 2), chunks=(5, 2))
+    frame = dd.from_dask_array(arr, columns=['a', 'a'])
+    repr(frame)
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index 6f1c0a01b..24cef408b 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -19,6 +19,7 @@ from dask.delayed import Delayed
 from dask.utils import tmpdir, tmpfile, ignoring
 from dask.utils_test import inc, dec
 from dask.compatibility import long, unicode, PY2
+from dask.diagnostics import Profiler
 
 
 def import_or_none(path):
@@ -868,3 +869,18 @@ def test_callable_scheduler():
 
     assert delayed(lambda: 1)().compute(scheduler=get) == 1
     assert called[0]
+
+
+@pytest.mark.skipif('not da')
+@pytest.mark.parametrize('num_workers', range(1, 4))
+@pytest.mark.parametrize('scheduler', ['threads', 'processes'])
+def test_num_workers_config(num_workers, scheduler):
+    # Regression test for issue #4082
+    a = da.random.randint(0, 10, size=400, chunks=2)
+
+    with dask.config.set(num_workers=num_workers), Profiler() as prof:
+        a.compute(scheduler=scheduler)
+
+    workers = {i.worker_id for i in prof.results}
+
+    assert len(workers) == num_workers