[SPARK-2538] [PySpark] Hash based disk spilling aggregation

During aggregation in Python worker, if the memory usage is above spark.executor.memory, it will do disk spilling aggregation. 

It will split the aggregation into multiple stage, in each stage, it will partition the aggregated data by hash and dump them into disks. After all the data are aggregated, it will merge all the stages together (partition by partition).
When objects are freed in Python, the memory may be not released to system immediately. For example, given limit=512M, but the used memory goes up to 550M until it find out this. After spilling, the used memory is still 520M (300MB of them is free in Python). If we still use limit=512M, then after merging 1000 objects, it will dump the data into disks. If we use limit=530M, then it will dump the objects until them occupied all the free memory in Python. The later approach will have better performance.

This will be especially useful when people have small spark.python.work.memory but huge memory in the machine. 

How do you think of it?