Does spark operator support dynamic resource allocation?

[paulpaul1076]

I haven't seen any mention of dynamic resource allocation for spark jobs in stackable docs. Is it supported? Or is it more of a "use it at your own risk" type of thing?

Thanks.

[adwk67]

It is supported partially, in the sense that you can set min/max CPU and a memory limit for the executors defined for your SparkApplication. What we don't support is a dynamic number of executors. We may do in the future but there are a number of factors that need to be considered:

• the Kubernetes-specific Spark issue is still open and I think we would prefer to wait until this has been completed. See some of the provisos mentioned here.
• the Spark issue (see also the PR) that enhances shuffle-tracking allows an initial number if executors to be set, but then things can complicated when workflow distribution needs to accomodate other settings such as affinities etc.

So...the k8s "paradigm" moves away from a cluster running a static number of nodes to pods-on-demand, and resources for these pods can be set on a per-Job basis: is that sufficient for your use-cases? If not, what specific scenario are you looking at that could be improved by dynamic allocation?

[paulpaul1076]

@adwk67 is there an example of how to set min/max CPU and memory limit? On the internet I've only seen examples of min/max executor numbers. So, I wouldn't even know the name of the setting.

You say that dynamic number of executors isn't supported, but yesterday I literally tried setting spark.dynamicAllocation.minExecutors and spark.dynamicAllocation.maxExecutors and I saw the number of executors increase and decrease. How is this possible when you say that it isn't supported?

These are the settings I set:

    spark.shuffle.service.enabled: 'false'
    spark.dynamicAllocation.enabled: 'true'
    spark.dynamicAllocation.executorIdleTimeout: '5'
    spark.dynamicAllocation.shuffleTracking.enabled: 'true'
    spark.dynamicAllocation.initialExecutors: '1'
    spark.dynamicAllocation.minExecutors: '0'
    spark.dynamicAllocation.maxExecutors: '7'

You can see the screenshot below, it shows that the number of executors would change as my spark application was running. (I was checking it with kubectl get pods)

[adwk67]

Thanks for your reply and the details you share. Sorry, I wasn't clear: dynamic pods are not officially supported by us, even if it works by passing spark configuration settings straight through to the spark-submit command (which is what we do with things listed under sparkConf). For simple cases this may be fine but we have not tested it with other features such as Pod availability, Pod disruption budgets and affinities and so can't (at this time) give any guidelines or guarantees.

Regarding setting resources: you can look at this integration test to see how they are specified and what is expected.

[paulpaul1076]

Thanks.

I see spark.kubernetes.driver.request.cores, spark.kubernetes.driver.limit.cores, spark.kubernetes.executor.request.cores, spark.kubernetes.executor.limit.cores. I guess that's the way you set min/max for cores?

As for memory min/max, I saw this:

spec:
  containers:
    - name: spark-submit
      resources:
        limits:
          cpu: 400m
          memory: 512Mi
        requests:
          cpu: 100m
          memory: 512Mi

But it also specifies cores, too. So, which one will eventually be used? The one from Spark config, or this one? What's the priority?

In the yaml files I've used for submitting spark applications I also have this:

  driver:
    resources:
      cpu:
        min: '250m'
        max: "1"
      memory:
        limit: '1Gi'
    volumeMounts:
      - name: job-deps
        mountPath: /dependencies/jars
  executor:
    instances: 4
    resources:
      cpu:
        min: '250m'
        max: "1"
      memory:
        limit: '1Gi'
    volumeMounts:
      - name: job-deps
        mountPath: /dependencies/jars

I guess it's yet another place where this can be specified? I can see the min/max for cpu here, but how would I specify the min/max for memory? I can see limit there, I guess limit=max? How would I specify min? Through the request keyword?

Btw, do configs like spark.kubernetes.driver.request.cores support the 400m syntax? Or can only natural numbers be used there?

[adwk67]

Have a look at the nightly docs as this should make things a little clearer: https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/resources
As noted there, sparkConf properties take precedence:

It is possible to define Spark resources either directly by setting configuration properties listed under sparkConf, or by using resource limits. If both are used, then sparkConf properties take precedence. It is recommended for the sake of clarity to use either one or the other. See below for examples.

See also https://docs.stackable.tech/home/nightly/concepts/resources and https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/#resource-units-in-kubernetes for information on the units used by Kubernetes and which we (try to!) reflect in our resource definitions.

[adwk67]

And, yes, the current implementation is that for memory, the limit is used for both request and limit.