Spark-on-k8s-operator: Spark Dynamic Automatic scaling issue

[vishnugs]

Hi ,
Spark Version is 2.4.4

I have enabled the following configuration on the spark yaml applicaiton. But it fails to start as executors keeps getting killed with exit code 1.

 "spark.dynamicAllocation.enabled": "true"
 "spark.dynamicAllocation.initialExecutors": "1"
 "spark.dynamicAllocation.minExecutors": "2"
 "spark.dynamicAllocation.maxExecutors": "3"
 "spark.dynamicAllocation.executorIdleTimeout": "7000s"
 "spark.shuffle.service.enabled": "true"
 "spark.dynamicAllocation.shuffleTracking.enabled": "true"
 "spark.network.timeout": "700s"
 "spark.shuffle.registration.timeout": "6000s"

Lost executor 1 on 10.42.5.73: Unable to create executor due to Unable to register with external shuffle server due to : Failed to connect to /10.42.5.73:7337

spark-driver-k8s.log

Regards
Vishnu

[vishnugs]

Executor logs

2020-01-02 09:28:12 ERROR Logging$class:91 Failed to connect to external shuffle server, will retry 1 more times after waiting 5 seconds...
java.io.IOException: Failed to connect to /10.42.5.144:7337
at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:245)
at org.apache.spark.network.client.TransportClientFactory.createUnmanagedClient(TransportClientFactory.java:201)
at org.apache.spark.network.shuffle.ExternalShuffleClient.registerWithShuffleServer(ExternalShuffleClient.java:142)
at org.apache.spark.storage.BlockManager$$anonfun$registerWithExternalShuffleServer$1.apply$mcVI$sp(BlockManager.scala:294)
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)
at org.apache.spark.storage.BlockManager.registerWithExternalShuffleServer(BlockManager.scala:291)
at org.apache.spark.storage.BlockManager.initialize(BlockManager.scala:265)
at org.apache.spark.executor.Executor.(Executor.scala:118)
at org.apache.spark.executor.CoarseGrainedExecutorBackend$$anonfun$receive$1.applyOrElse(CoarseGrainedExecutorBackend.scala:83)
at org.apache.spark.rpc.netty.Inbox$$anonfun$process$1.apply$mcV$sp(Inbox.scala:117)
at org.apache.spark.rpc.netty.Inbox.safelyCall(Inbox.scala:205)
at org.apache.spark.rpc.netty.Inbox.process(Inbox.scala:101)
at org.apache.spark.rpc.netty.Dispatcher$MessageLoop.run(Dispatcher.scala:221)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /10.42.5.144:7337
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:323)
at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:340)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:633)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:858)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:138)
... 1 more
Caused by: java.net.ConnectException: Connection refused
... 11 more

[liyinan926]

Spark on Kubernetes doesn't support dynamic resource allocation yet. There's no shuffle service available for the Kubernetes mode yet.

[vishnugs]

When is the plan to support this ? Any plan in upcoming release ?

[chiragkumarp]

@liyinan926 as Vishnu's question, we are curious to know about the availability of dynamic resources allocation capability. Is this in current roadmap if so then what is the ETA? If we are using mode as a cluster, does this mean K8s local resource manager able to distribute workloads among number of executor automatically by Spark operator (Driver pod) orchestration?

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
name: spark-test-app
namespace: default
spec:
type: Scala
mode: cluster

[sushiljacksparrow]

I am wondering if the comment made by liyinan926 is still true. My understanding is that it is supported if you use spark-3.0

[liyinan926]

@sushiljacksparrow what is supported in Spark 3.0 is a form of dynamic resource allocation that is different than what you get in other scheduler backends. See https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/issues/616. BTW: we now have the following 3.0.0 images:

gcr.io/spark-operator/spark:v3.0.0
gcr.io/spark-operator/spark-py:v3.0.0
gcr.io/spark-operator/spark-operator:v1beta2-1.1.2-3.0.0

[liyinan926]

Dynamic allocation via shuffle tracking will be supported in https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/pull/976.

[vishnugs]

@liyinan926 is the feature live now with dynamic scaling out, scaling in of executors ?