[BEAM-1074] Set default-partitioner in SourceRDD.Unbounded.

• Type: Improvement
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Fix Version/s: 2.0.0
• Component/s: runner-spark
• Assignee: Aviem Zur
• Reporter: Amit Sela
• Created: 02/Dec/16 17:06
• Updated: 16/May/20 13:45
• Resolved: 26/Mar/17 03:53

Description

The SparkRunner uses mapWithState to read and manage CheckpointMarks, and this stateful operation will be followed by a shuffle:
https://github.com/apache/spark/blob/master/streaming/src/main/scala/org/apache/spark/streaming/dstream/MapWithStateDStream.scala#L159

Since the stateful read maps "splitSource" -> "partition of a list of read values", the following shuffle won't benefit in any way (the list of read values has not been flatMapped yet). In order to avoid shuffle we need to set the input RDD (SourceRDD.Unbounded) partitioner to be a default HashPartitioner since mapWithState would use the same partitioner and will skip shuffle if the partitioners match.

Issue Links

• relates to: BEAM-848 Shuffle input read-values to get maximum parallelism.
• links to: GitHub Pull Request #1500
• links to: GitHub Pull Request #2288