[SPARK-32734] RDD actions in DStream.transfrom delays batch submission

• Type: Bug
• Status: In Progress
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 3.0.0
• Component/s: DStreams
• Docs Text: Fix batch submission delay caused by RDD actions in DStream.transfrom
• Assignee: Unassigned
• Reporter: Liechuan Ou
• Created: 29/Aug/20 14:28
• Updated: 03/Sep/20 00:04

Description

Issue

Some spark applications have batch creation delay after running for some time. For instance, Batch 10:03 is submitted at 10:06. In spark UI, the latest batch doesn't match current time.
 

Clock	BatchTime
10:00	10:00
10:02	10:01
10:04	10:02
10:06	10:03

Investigation

We observe such applications share a commonality that rdd actions exist in dstream.transfrom. Those actions will be executed in dstream.compute, which is called by JobGenerator. JobGenerator runs with a single thread event loop so any synchronized operations will block event processing.

Proposal

delegate dstream.compute to JobSchduler

 

// class ForEachDStream

override def generateJob(time: Time): Option[Job] = {
  val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
    parent.getOrCompute(time).foreach(rdd => foreachFunc(rdd, time))
  }
  Some(new Job(time, jobFunc))
}

Issue Links

• links to: [Github] Pull Request #29578 (Olwn)