[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).
I see, could we force GC instead with gc.collect()? That would be cleaner if it worked. This approach seems kind of troublesome, e.g. we might ramp up to much more than the limit if we're not careful.

Another option is to stop checking for more than 1000 items. You can estimate the number of items after which you should begin checking again by looking a how many items you processed until the first time you had to spill.