Hello, Habr! Yesterday at the , hosted by the guys from Rambler&Co, there were quite a few questions from participants related to configuring this tool. We decided to share our experience following up. It's a complex topic — so we encourage sharing experiences in the comments as well; perhaps we also misunderstand or misuse something.
A brief introduction — how we use Spark. We have a three-month program , and in the second module, our participants work with this tool. Accordingly, our task as organizers is to prepare a cluster for use within such a case.
The peculiarity of our use is that the number of people working on Spark simultaneously can be equal to the entire group. For instance, during a seminar, when everyone is trying out something together and following our instructor. That can be quite a lot — sometimes nearly 40 people. There probably aren't many companies in the world that face such a usage scenario.
Next, I'll explain how and why we selected certain configuration parameters.
Let’s start from the very beginning. Spark has three options for running on the cluster: standalone, using Mesos, and using YARN. We decided to choose the third option, as it made sense for us. We already have a Hadoop cluster, and our participants are well acquainted with its architecture. Let’s use YARN.
spark.master=yarnNext, it gets more interesting. Each of these three deployment options has two deployment modes: client and cluster. Based on various online references, one can conclude that the client mode is suitable for interactive work — for example, through Jupyter Notebook, while the cluster mode is more suitable for production solutions. In our case, we were interested in interactive work, so:
spark.deploy-mode=clientFrom this moment, Spark will begin functioning on YARN, but that wasn't enough for us. Since our program is focused on big data, sometimes participants lacked the resources provided by the uniform allocation. Here, we discovered an interesting feature — dynamic resource allocation. In short, the essence is as follows: if you have a heavy task and the cluster is free (for instance, in the morning), Spark can allocate additional resources for you with this option. The necessity is determined by a clever formula. We won't delve into the details — it works quite well.
spark.dynamicAllocation.enabled=trueWe set this parameter, and upon starting Spark, it complained and didn't launch. Rightly so, because we needed to read more carefully. It states that to ensure everything is okay, an additional parameter must also be enabled.
spark.shuffle.service.enabled=trueWhat is it for? When our job no longer requires such a large amount of resources, Spark should return them to the common pool. The most resource-intensive stage in almost any MapReduce job is the Shuffle stage. This parameter allows saving the data generated at this stage and, accordingly, freeing up executors. An executor is the process that performs all the calculations on the worker. It has a certain number of CPU cores and a certain amount of memory.
We added this parameter. Everything seemed to work. It became apparent that participants were indeed receiving more resources when they needed them. But another problem arose — at some point, other participants woke up and also wanted to use Spark, but everything was occupied, and they were unhappy. It's understandable. We started looking into the documentation. It turned out there are additional parameters that can influence the process. For example, if an executor is in idle mode — after how long can resources be reclaimed?
spark.dynamicAllocation.executorIdleTimeout=120sIn our case, if your executors do nothing for two minutes, please return them to the general pool. However, this parameter was not always sufficient. It was clear that a person had been inactive for a long time, yet resources were not being released. It turned out that there was a special parameter — how long to wait before reclaiming executors that contain cached data. By default, this parameter was set to infinity! We corrected it.
spark.dynamicAllocation.cachedExecutorIdleTimeout=600sSo if your executors do nothing for 5 minutes, return them to the general pool. In this mode, the speed of resource release and allocation for a large number of users became satisfactory. The level of dissatisfaction decreased. However, we decided to go further and limit the maximum number of executors per application — essentially per participant in the program.
spark.dynamicAllocation.maxExecutors=19Of course, now some were unhappy from the other side — "the cluster is idle, and I only have 19 executors," but what can you do — a proper balance is needed. It is impossible to make everyone happy.
And one more small story related to the specifics of our case. A few people were late for a practical lesson, and for some reason, their Spark did not start. We looked at the number of free resources — there seemed to be enough. Spark should start. Fortunately, by that time the documentation was already ingrained in our minds, and we remembered that when starting Spark looks for a port to start on. If the first port in the range is occupied, it moves to the next one. If it’s free, it takes it. There is a parameter that specifies the maximum number of attempts for this. By default, it is 16. This is fewer than the number of people in our class. Accordingly, after 16 attempts, Spark would give up and say that it could not start. We adjusted this parameter.
spark.port.maxRetries=50Next, I will tell you about some settings that are not closely related to the specifics of our case.
For a faster Spark startup, it is recommended to archive the jars folder located in the home directory SPARK_HOME and place it on HDFS. This way, it will not waste time loading these jars on workers.
spark.yarn.archive=hdfs:///tmp/spark-archive.zipTo achieve better performance, it's recommended to use Kryo as the serializer. It's more optimized than the default one.
spark.serializer=org.apache.spark.serializer.KryoSerializerThere's also a long-standing issue with Spark where it often runs out of memory. This usually happens when workers have calculated everything and are sending results to the driver. We increased this parameter. By default, it's 1GB; we set it to 3GB.
spark.driver.maxResultSize=3072Lastly, as a bonus. How to update Spark to version 2.1 on the HortonWorks distribution — HDP 2.5.3.0. This version of HDP comes with a pre-installed version 2.0, but we decided some time ago that Spark is rapidly evolving, and each new version fixes bugs and provides additional capabilities, including for the Python API, so we decided to do the update.
We downloaded the version from the official website for Hadoop 2.7. Unzipped it and placed it in the HDP folder. We created symlinks as needed. When we tried to run it — it doesn't start. It shows a very unclear error.
java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfigAfter some research, we found out that Spark decided not to wait for Hadoop to release its updates, opting to use a new version of Jersey. They argue about this in JIRA. The solution was to download . Place this in the jars folder in SPARK_HOME, zip it again, and place it on HDFS.
We bypassed this error, but a new and somewhat vague one appeared.
org.apache.spark.SparkException: Yarn application has already ended! It might have been killed or unable to launch application master.Meanwhile, we tried running version 2.0 — it's all fine. Try to guess what the issue is. We checked the logs of this application and saw something like this:
/usr/hdp/${hdp.version}/hadoop/lib/hadoop-lzo-0.6.0.${hdp.version}.jarIn general, for some reason, hdp.version did not resolve. After some research, we found the solution. We need to go into Ambari's settings for YARN and add a parameter to the custom yarn-site:
hdp.version=2.5.3.0-37This magic helped, and Spark took off. We tested several of our Jupyter notebooks. Everything works. We are ready for the first Spark class on Saturday (tomorrow)!
UPD. During the class, another problem arose. At some point, YARN stopped providing containers for Spark. We needed to adjust a parameter in YARN that was set to 0.2 by default:
yarn.scheduler.capacity.maximum-am-resource-percent=0.8 That is, only 20% of the resources were involved in resource allocation. After changing the parameters, we restarted YARN. The issue was resolved, and the other participants were also able to start the Spark context.
Source: habr.com
