[SPARK-1683] Track task read metrics.

This commit adds a new metric in TaskMetrics to record
the input data size and displays this information in the UI.

An earlier version of this commit also added the read time,
which can be useful for diagnosing straggler problems,
but unfortunately that change introduced a significant performance
regression for jobs that don't do much computation. In order to
track read time, we'll need to do sampling.

The screenshots below show the UI with the new "Input" field,
which I added to the stage summary page, the executor summary page,
and the per-stage page.

![image](https://cloud.githubusercontent.com/assets/1108612/3167930/2627f92a-eb77-11e3-861c-98ea5bb7a1a2.png)

![image](https://cloud.githubusercontent.com/assets/1108612/3167936/475a889c-eb77-11e3-9706-f11c48751f17.png)

![image](https://cloud.githubusercontent.com/assets/1108612/3167948/80ebcf12-eb77-11e3-87ed-349fce6a770c.png)
How can a talk end up reading data from both HDFS and the cache? I didn't
realize that was possible.


On Wed, Jun 25, 2014 at 2:05 PM, andrewor14 <notifications@github.com>
wrote:

> In core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala:
>
> > @@ -67,6 +67,12 @@ class TaskMetrics extends Serializable {
> >    var diskBytesSpilled: Long = _
> >
> >    /**
> > +   * If this task reads from a HadoopRDD, from cached data, or from a parallelized collection,
>
> I see. What if the same task reads the data from both HDFS and the cache?
> From the code it seems that we keep the input bytes from the cache and
> overwrite the ones for HDFS. Maybe I'm misunderstanding but I don't see an
> easy way to tell whether these bytes are for an external source or from the
> cache.
>
> —
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/962/files#r14212270>.
>