Consider automatic keep alives

[johnynek]

In a large flatMap expansion, we have a loop like:

fn(input).foreach { it =>
  collector.add(it)
}

if the result of fn(input) is large and takes a long time to create each item, we could get a timeout as we do this. It appears that adding to a collector does not send the keep alive to the job tracker. We could keep a counter and do something like:

var count = 0
fn(input).foreach { it =>
  collector.add(it)
  count += 1
  if(count % SEND_EVERY == 0) {
    flowProcess.keepAlive
  }
}

We could do the same in the reducers when we apply the Iterator => Iterator function and enumerate that.