[SPARK-15480][UI][Streaming]show missed InputInfo in streaming UI

[mwws]

What changes were proposed in this pull request?

It's a bug in Streaming UI. If BatchDuration is changed by window operation, InputInfo on time with no output ops will not be shown in streaming UI.

Here is a simple example to reproduce:

val inputDStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParam, topic)
val matchedDStream = inputDStream.window(batchInteval*2, batchInteval*2)
matchedDStream.foreachRDD(rdd => {
    val count = rdd.count()
    println(s"2 count number: ${count}")
})

How was this patch tested?

manually tested

I manually input 6 records from Kafka. According to output, there are indeed 6 records processed

but according to web UI, input records show 0.

And here is the Screenshot after my change:

[SparkQA]

Test build #59124 has finished for PR 13259 at commit 83351b8.

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

[mwws]

@chenghao-intel @carsonwang

[chenghao-intel]

cc @zsxwing

[jerryshao]

Will this change the semantics if posing fake StreamingListenerBatchCompleted, if user's code rely on this, will this break their assumptions?

[mwws]

Good point, it would be a breaking change in that case. Give the fact that some information is indeed missing now, we need either send more events or add additional fields in current event. The later might be better and more correct in semantics, although it can still break current implementation of user's custom listener. And I noticed that listener interface is annotated as DeveloperAPI, so it's expected that API might be changed. How do you think?

[zsxwing]

I suggest we change InputInfoTracker.getInfo(batchTime: Time) to InputInfoTracker.getInfo(batchTime: Time, batchDuration: Long) and let InputInfoTracker return the aggregated input infos between batchTime - batchDuration and batchTime.

[mwws]

@zsxwing thank you for the suggestion, But I have two concerns:

1. There could be multiple output ops on one BatchTime. And these ops could have different batchDuration. Which one should I use?
2. The input rate graph is generated by the input size column of batch table. If we do the aggregation, the meaning of input rate graph will be changed from "number of input events from source" to "number of processed events".

[zsxwing]

@zsxwing thank you for the suggestion, But I have two concerns:

1. There could be multiple output ops on one BatchTime. And these ops could have different batchDuration. Which one should I use?
2. The input rate graph is generated by the input size column of batch table. If we do the aggregation, the meaning of input rate graph will be changed from "number of input events from source" to "number of processed events".

How about adding a field lastBatchTime in JobScheduler and storing the last submitted batch? Then just use the aggregated input infos between the last batch and the current batch? We need to aggregate them since we only have batch infos that have jobs.

[mwws]

@zsxwing Thanks for your suggestion, I have re-implemented this PR according to that. But instead of maintaining state lastBatchTime in JobScheduler, I would prefer record such info in JobGenerator, so we can have central place to resolve inputInfo.

I according to my manual test the new code works. Could you help review again?

[SparkQA]

Test build #59730 has finished for PR 13259 at commit a3d05e0.

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

[zsxwing]

@mwws sorry for the delay. Could you add a unit test, please?

[mwws]

@zsxwing Yes sir, a unit test has been added.

[SparkQA]

Test build #60095 has finished for PR 13259 at commit ffd1787.

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

[tdas]

@zsxwing Could you take a look at this PR. Would be good to merge this.

[mwws]

@zsxwing This patch has been pended for a long time, could you help to review?