Skip to content

Conversation

@mayuehappy
Copy link

In SparkStreaming DRA .The metric we use to add or remove executor is the ratio of batch processing time / batch duration (R). And we use the parameter "spark.streaming.dynamicAllocation.maxExecutors" to set the max Num of executor .Currently it doesn't work well with Spark streaming because of several reasons:
(1) For example if the max nums of executor we need is 10 and we set "spark.streaming.dynamicAllocation.maxExecutors" to 15,Obviously ,We wasted 5 executors.
(2) If the number of topic partition changes ,then the partition of KafkaRDD or the num of tasks in a stage changes too.And the max executor we need will also change,so the num of maxExecutors should change with the nums of Task .
The goal of this JIRA is to auto fix maxNumExecutors . Using a SparkListerner when Stage Submitted ,first figure out the num executor we need , then update the maxNumExecutor

@jerryshao
Copy link
Contributor

IIUC streaming DRA seems an obsolete code. Long ago when I played with it, there existed some bugs, but seems not so many users used this feature. I'm not sure if we really need to put efforts on this code, since we now move to Structured Streaming instead.

@AmplabJenkins
Copy link

Can one of the admins verify this patch?

@srowen
Copy link
Member

srowen commented Jul 2, 2018

(There are a few minor style problems in the code -- watch spacing.)
I am not clear that there is a problem here. Setting max executors to 15 and using 10 isn't a problem. If number of executors that could be used increases, it also doesn't mean the max must accommodate it, or else what is the point of the max? this would need a clearer explanation.

@mayuehappy
Copy link
Author

@srowen Really thanks for your reply.I think maybe I didn't express it well.Let's assume that there is a situation like this. If we use spark streaming to consume a Kafka topic with 10 partition, so it will make a KafkaRDD with 10 partition.that also means the number of task is 10. if our each executor core is 1, concurrentJob = 1,can we consider that there can only use a maximum of 10 executor at same time.If we turned on the Dynamic allocation of resources and set maxExecutors to 15 .Now Spark Streaming's resource allocation strategy is using the rate ProcessTime/BatchSize,if our ProcessTime/BatchSize > scalingUpRatio,the job will acontinues to apply resources from yarn ,but these new executor are useless. So I think we should limit the MaxExecutor = numTasks * conCurrentJobs * cpuPerTask / coresPerExecutor.I don't know if I have a clear expression = =。

@srowen
Copy link
Member

srowen commented Jul 5, 2018

If you know you won't need more than 10, then set the max to 10.

If you don't necessarily know that, then I think you're complaining that dynamic allocation doesn't 'know' how many executors will be needed in advance. Yes, in general the load goes up and down and can't be predicted, so dynamic allocation is always adapting, and will add executors or time out idle ones eventually to match load. This is just how it works.

I think you're suggesting a specific strategy for Spark Streaming jobs only. While I understand it, because you do know more about the load in this type of job, this is also a reason to set the max because you know what it should be, or simply not use Spark Streaming. It's often not used in streaming because the lag of adapting to a new load of tasks increases latency and variability.

Just set your max to 10, or perhaps set it to more rapidly time out idle executors.

@srowen srowen mentioned this pull request Jul 18, 2018
@asfgit asfgit closed this in 1a4fda8 Jul 19, 2018
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants