title: Dask bag gets stuck on processing when blocksize is defined, only uses one worker when it's not defined
body: <p>I'm trying to process a single large (1TB) json file locally with Dask. The file has one object per line. When I don't specify blocksize in the <code>read_text</code> function, the code runs perfectly but only on one worker. Only one partition is then made, and only one task can be seen in the dashboard. If I do specify <code>blocksize</code>, all workers get tasks, but they never move on from processing (at least not within 12 hours). What is going wrong? How do I get all workers to actually do work?</p>
  
  <p>The code is as follows:</p>
  
  <pre class="lang-py prettyprint-override"><code>import dask.bag as db
  from dask.distributed import Client, LocalCluster
  cluster = LocalCluster(n_workers=4, 
                  threads_per_worker=2,
                  memory_limit='2GB')
  
  client = Client(cluster)
  
  db.read_text('./data/uncompressed/latest-all.json', blocksize=1e8)\
      .map(lambda obj: obj[:-2])\
      .map(parse_json)\
      .map(prune)\
      .filter(lambda obj: obj != None)\
      .map(json.dumps)\
      .to_textfiles('./data/proc/*.json')
  
  </code></pre>
  
  <p>The <code>parse_json</code> and <code>prune</code> are both pure python functions with no IO.</p>
  
  <p>Here's a snip of the dashboard when blocksize is defined:
  <a href="https://i.stack.imgur.com/Vb1H4.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/Vb1H4.png" alt="dask dashboard"></a></p>