[SPARK-30667] Support simple all gather in barrier task context

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Done
• Affects Version/s: 3.0.0
• Component/s: PySpark, Spark Core
• Target Version/s: 3.0.0
• Assignee: Sarth Frey
• Reporter: Xiangrui Meng
• Created: 28/Jan/20 19:30
• Updated: 21/May/20 08:09
• Resolved: 10/Mar/20 21:29

Description

Currently we offer task.barrier() to coordinate tasks in barrier mode. Tasks can see all IP addresses from BarrierTaskContext. It would be simpler to integrate with distributed frameworks like TensorFlow DistributionStrategy if we provide all gather that can let tasks share additional information with others, e.g., an available port.

Note that with all gather, tasks are share their IP addresses as well.

port = ... # get an available port
ports = context.all_gather(port) # get all available ports, ordered by task ID
...  # set up distributed training service

Issue Links

• causes: SPARK-31784 Fix test BarrierTaskContextSuite."share messages with allGather() call"
• links to: GitHub Pull Request #27395
• links to: GitHub Pull Request #27640
• links to: GitHub Pull Request #27951