title: Dask - Drop duplicate index MemoryError
body: <p>I'm getting a <code>MemoryError</code> when I try to drop duplicate timestamps on a large dataframe with the following code.</p>
  
  <pre><code>import dask.dataframe as dd
  
  path = f's3://{container_name}/*'
  ddf = dd.read_parquet(path, storage_options=opts, engine='fastparquet')
  ddf = ddf.reset_index().drop_duplicates(subset='timestamp_utc').set_index('timestamp_utc')
  ...
  </code></pre>
  
  <p>Profiling shows that it was using up about 14GB of RAM on a dataset of 265MB of gzipped parquet files containing about 40 million rows of data.</p>
  
  <p>Is there an alternative way I can drop duplicate indexes on my data without Dask using so much memory?</p>
  
  <p>The traceback below</p>
  
  <pre><code>Traceback (most recent call last):
    File "/anaconda/envs/surb/lib/python3.6/runpy.py", line 193, in _run_module_as_main
      "__main__", mod_spec)
    File "/anaconda/envs/surb/lib/python3.6/runpy.py", line 85, in _run_code
      exec(code, run_globals)
    File "/home/chengkai/surbana_lift/src/consolidate_data.py", line 62, in &lt;module&gt;
      consolidate_data()
    File "/home/chengkai/surbana_lift/src/consolidate_data.py", line 37, in consolidate_data
      ddf = ddf.reset_index().drop_duplicates(subset='timestamp_utc').set_index('timestamp_utc')
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/dataframe/core.py", line 2524, in set_index
      divisions=divisions, **kwargs)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/dataframe/shuffle.py", line 64, in set_index
      divisions, sizes, mins, maxes = base.compute(divisions, sizes, mins, maxes)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/base.py", line 407, in compute
      results = get(dsk, keys, **kwargs)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/threaded.py", line 75, in get
      pack_exception=pack_exception, **kwargs)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/local.py", line 521, in get_async
      raise_exception(exc, tb)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/compatibility.py", line 67, in reraise
      raise exc
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/local.py", line 290, in execute_task
      result = _execute_task(task, data)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/local.py", line 270, in _execute_task
      args2 = [_execute_task(a, cache) for a in args]
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/local.py", line 270, in &lt;listcomp&gt;
      args2 = [_execute_task(a, cache) for a in args]
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/local.py", line 267, in _execute_task
      return [_execute_task(a, cache) for a in arg]
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/local.py", line 267, in &lt;listcomp&gt;
      return [_execute_task(a, cache) for a in arg]
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/local.py", line 271, in _execute_task
      return func(*args2)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/dataframe/core.py", line 69, in _concat
      return args[0] if not args2 else methods.concat(args2, uniform=True)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/dask/dataframe/methods.py", line 329, in concat
      out = pd.concat(dfs3, join=join)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/pandas/core/reshape/concat.py", line 226, in concat
      return op.get_result()
    File "/anaconda/envs/surb/lib/python3.6/site-packages/pandas/core/reshape/concat.py", line 423, in get_result
      copy=self.copy)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/pandas/core/internals.py", line 5418, in concatenate_block_manage
  rs
      [ju.block for ju in join_units], placement=placement)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/pandas/core/internals.py", line 2984, in concat_same_type
      axis=self.ndim - 1)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/pandas/core/dtypes/concat.py", line 461, in _concat_datetime
      return _concat_datetimetz(to_concat)
    File "/anaconda/envs/surb/lib/python3.6/site-packages/pandas/core/dtypes/concat.py", line 506, in _concat_datetimetz
      new_values = np.concatenate([x.asi8 for x in to_concat])
  MemoryError
  </code></pre>