Spark on EGO in Platform Conductor

Body

Apache Spark is a powerful open source processing engine built around speed, ease of use, and sophisticated analytics. Apache Spark applications run as independent sets of processes on a cluster, coordinated by the SparkContext object in the driver program. SparkContext can connect to the cluster manager (either Spark’s own standalone cluster manager, Mesos, or YARN), which allocates resources across applications. With Platform Conductor for Spark (Platform Conductor), Platform Enterprise Grid Orchestrator (EGO) acts as the cluster manager with Apache Spark, enabling Spark applications to benefit from resource sharing across the cluster.

EGO provides a full suite of services to support and manage resource orchestration in a cluster including cluster management, configuration and auditing of service-level plans, resource facilitation to provide failover if a master host goes down, monitoring, and data distribution. EGO is sensitive only to the resource requirements of business services, and has no knowledge of any runtime dynamic parameters that exist for it. This means that EGO does not interfere with how a business service chooses to use the resources it has been allocated.

The following image represents the Spark on EGO architecture:

Spark on EGO allows EGO to serve as the cluster manager, and provides the following benefits:

Fine-grained scheduling
Apache Spark uses a coarse-grained or “fine-grained” resource scheduling policy. The Spark application requests a static number of resources and holds them in its lifecycle, which means each application gets more or fewer resources as it scales up and down. Based on the fine-grained scheduling policy, applications share resources at a very fine granularity, especially many applications running in a cluster concurrently.

Resource Reclaim
Tenants (known as a Spark instance group in Platform Conductor) share resources in a Spark on EGO cluster, which means some tenants can acquire more resources than the resource distribution definition by borrowing them from other applications. When the lender needs more resources, it will reclaim these borrowed resources from the borrower. This can keep the cluster in high usage status and maintain consistency between all tenants.

Resources acquired concurrently
EGO permits all clients to send resource requests together and partition the resources based on the resource allocation plan.

Session scheduler
A consumer can run many applications concurrently, and the session scheduler can schedule resources between applications belonging to one consumer. This means the resource scheduling in Spark on EGO is hierarchical. The following image describes this hierarchy:

Frequently used parameters:
SPARK_EGO_EXECUTOR_SLOTS_MAX=<max_value> / spark.ego.executor.slots.max=<max_value>
This parameter can specify the maximum number of tasks for each executor, and is set to the maximum value by default. When the executor is disconnected from the driver because it is “out of memory”, this means one executor is running too many tasks at the same time. You should evaluate an executor’s memory and memory usage per task to choose the proper value and setting for SPARK_EGO_EXECUTOR_SLOTS_MAX to decrease the concurrent number of tasks running on the executor.

SPARK_EGO_SLOTS_MAX=<max_slots> / spark.ego.slots.max=<max_value>
This parameter specifies the maximum number of slots that an application can receive. For example, if each task can be completed in 200ms and the total number of tasks is 200, then the application needs about 43s (200ms * 200 + 3s for communication) to complete all tasks with one computing slot. Setting the SPARK_EGO_SLOTS_MAX to a suitable value can decrease an application’s run time to a reasonable range (10 slots > 7s, 20 slots > 5s) and economize resource usage in the cluster.

SPARK_EGO_EXECUTOR_SLOTS_RESERVE=<reserve_value> / spark.ego.executor.slots.reserve=<reserve_value>
Spark on EGO allocates resources based on an application’s requirement. For one Spark application, which can contains many stages, the resource requirement for each stage is different. If the requirement for one stage is smaller than the last one, the driver will release the unused resources back to session scheduler. If the requirement for one stage is larger than the last one, the driver will request more resources, which will require more time to fetch the new resources. You can increase the SPARK_EGO_EXECUTOR_SLOTS_RESERVE value to tell the driver to retain reserve_value resources, but refrain from setting this value too high as you may be inefficiently using resources that can be allocated to another application.

SPARK_EGO_EXECUTOR_IDLE_TIMEOUT=<timeout> / spark.ego.executor.idle.timeout=<timeout>
This parameter works with the SPARK_EGO_EXECUTOR_SLOTS_RESERVE parameter. When there are no running tasks on the application, each executor’s resources will be released except the reserve_value resources. If reserve_value=0, the executor will release all resources and shut itself down. If reserve_value > 0, the executor will shut itself down when the idle time is longer than the spark.ego.executor.idle.timeout parameter’s value. For spark-shell/Notebook, this value should be set to a larger number (approximately 600). When a new job is submitted, the executor can run tasks directly, and does not need to wait for newly starting executors. For batch jobs, you can use the default value of 60.

SPARK_EGO_PRIORITY =<number> / spark.ego.priority=<number>
This parameter specifies the priority of the application that is currently running. Some applications should be completed as soon as possible, and increasing the SPARK_EGO_PRIORITY value sets a higher priority over other applications to get more resources from the cluster. Note that when the master uses the FIFO policy, this parameter has no effect.

How to use these parameters:

Specify these parameters for a Spark instance group by editing the default Spark instance group configuration. Find these parameters under the settings for Spark on EGO and Session Scheduler.

You can also use the parameters as part of the spark-submit command. In this case, use the parameters in lowercase (for example, spark.ego.priority):

spark-submit ... --conf spark.ego.xxx=xxx --conf spark.ego.yyy=yyy ...

Now that you have an understanding of Spark on EGO in Platform Conductor, try it out. Download an evaluation version from our Service Management Connect page. If you have any questions, leave a comment in our forum!