[SPARK-3478] [PySpark] Profile the Python tasks

[davies]

This patch add profiling support for PySpark, it will show the profiling results
before the driver exits, here is one example:

============================================================
Profile of RDD<id=3>
============================================================
5146507 function calls (5146487 primitive calls) in 71.094 seconds
Ordered by: internal time, cumulative time
ncalls tottime percall cumtime percall filename:lineno(function)
5144576 68.331 0.000 68.331 0.000 statcounter.py:44(merge)
20 2.735 0.137 71.071 3.554 statcounter.py:33(__init__)
20 0.017 0.001 0.017 0.001 {cPickle.dumps}
1024 0.003 0.000 0.003 0.000 t.py:16(<lambda>)
20 0.001 0.000 0.001 0.000 {reduce}
21 0.001 0.000 0.001 0.000 {cPickle.loads}
20 0.001 0.000 0.001 0.000 copy_reg.py:95(_slotnames)
41 0.001 0.000 0.001 0.000 serializers.py:461(read_int)
40 0.001 0.000 0.002 0.000 serializers.py:179(_batched)
62 0.000 0.000 0.000 0.000 {method 'read' of 'file' objects}
20 0.000 0.000 71.072 3.554 rdd.py:863(<lambda>)
20 0.000 0.000 0.001 0.000 serializers.py:198(load_stream)
40/20 0.000 0.000 71.072 3.554 rdd.py:2093(pipeline_func)
41 0.000 0.000 0.002 0.000 serializers.py:130(load_stream)
40 0.000 0.000 71.072 1.777 rdd.py:304(func)
20 0.000 0.000 71.094 3.555 worker.py:82(process)

Also, use can show profile result manually by sc.show_profiles() or dump it into disk
by sc.dump_profiles(path), such as

>>>sc._conf.set("spark.python.profile", "true")
>>>rdd=sc.parallelize(range(100)).map(str)
>>>rdd.count()
100>>>sc.show_profiles()
============================================================ProfileofRDD<id=1>============================================================284functioncalls (276primitivecalls) in0.001secondsOrderedby: internaltime, cumulativetimencallstottimepercallcumtimepercallfilename:lineno(function)
40.0000.0000.0000.000serializers.py:198(load_stream)
40.0000.0000.0000.000 {reduce}
12/40.0000.0000.0010.000rdd.py:2092(pipeline_func)
40.0000.0000.0000.000 {cPickle.loads}
40.0000.0000.0000.000 {cPickle.dumps}
1040.0000.0000.0000.000rdd.py:852(<genexpr>)
80.0000.0000.0000.000serializers.py:461(read_int)
120.0000.0000.0000.000rdd.py:303(func)

The profiling is disabled by default, can be enabled by "spark.python.profile=true".

Also, users can dump the results into disks automatically for future analysis, by "spark.python.profile.dump=path_to_dump"

This is bugfix of #2351 cc @JoshRosen

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/20903/

[SparkQA]

QA tests have started for PR 2556 at commit e68df5a.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2556 at commit e68df5a.

• This patch passes unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[JoshRosen]

I've merged this. Thanks for the fix!