Apache Spark – How to create a powerful streaming application
After setting up Apache Kafka, the next step in our Retail Business Intelligence Platform project was to set up Spark, another widely used software solution from the Apache workshop. Apache Spark, the real spark necessary to ignite our project and turn it into a true stream processing arrangement, is probably the most famous streaming engine available in the modern IT world. Spark was initially developed at the University of California, Berkeley in 2009 as a research project and, as years went by, it has grown into an open-source project that is developed and maintained by a large and collaborative community. Infrastructure Just like all other components of this project, Spark has also been configured to run inside Amazon Elastic Compute Cloud. We have created two EC2 instances, both of m5a.2xlarge type: one that serves as a master node and one that serves as a worker node. It’s important to mention that the default amount of storage on the worker node wasn’t enough, we had to increase it to 40 GiB. Of course, the optimal amount of storage is different for each individual project and it’s often very hard to estimate it beforehand. We recommend starting with the default amount of storage and, if necessary, upgrading it later. Setup Before configuring Spark to run on EC2 instances, we had configured it for testing purposes by installing it locally, which was pretty simple and straightforward. It is enough to download and extract Spark to a folder, install Java and, given that we are writing our code in Python, install PySpark via pip package installer. After that, all that is left to do is to run the job using the following command: spark-submit –packages <NECESSARY_PACKAGE_NAMES_SEPARATED_BY_COLON> <PATH_TO_THE_PROGRAM> Back to the real stuff, let’s talk a little more about our in-cloud setup. First of all, when working with EC2 instances, one should be aware of the differences and possibilities each of the different AMIs comes with. For example, we explicitly needed to work with Python 3.9, but Amazon Linux 2 AMI, the default AMI for EC2 instances, at the moment of writing this blog post comes with Python 3.7 preinstalled. One would think: ‘Piece of cake, I just need to upgrade the Python version from 3.7 to 3.9, that’s a few minutes of work.’ Well, unfortunately, because of a lot of factors that aren’t related to the main theme of this blog post which is why they won’t be discussed any further, it turns out it isn’t that simple if wanted to be done properly. Although it is possible, we strongly recommend you, if in need of Python 3.9, simply choose another AMI. For instance, Red Hat AMI doesn’t come with Python pre-installed at all, which is why it’s simple to manually install whatever version is necessary (sudo yum install python39 in this case). Just as we installed Spark, Java, and PySpark on our local machine, we had to do it all again for Spark EC2 instances. Furthermore, it was necessary to generate an SSH key pair in order to use launch scripts in Spark’s sbin directory. The public key had to be copied to each worker node in the cluster. Finally, in order to use Spark’s launch scripts, we needed to edit the conf/slaves file on master node and fill in the workers’ hostnames. After everything mentioned is taken care of, the cluster should be started by running start-all.sh script in the sbin directory. However, as PySpark does not offer cluster mode, but only client mode, the job has to be executed in the client mode from one of the machines inside the cluster. We confirmed this with a little experiment. What we did was execute the job from our local PC and, as expected, the driver couldn’t communicate with the cluster. The thing is that the driver process runs on the machine that executed the spark-submit command. Generally, after the driver is started, it tries to communicate with nodes inside the cluster. Firstly with the master node in order to get an allocation of workers, and secondly with those workers. It is crucial that both master and workers can reach the driver, and vice versa. In this case, they couldn’t communicate with the driver that was being run inside our local network. The IP address shown in Figure 1 confirms that the driver was indeed inside our local network. Figure 1: Screenshot of a Spark Application UI To conclude, the job had to be run from either master or worker nodes in the client mode in order for everything to work smoothly. We had chosen to run it from the master node and that is why we installed Git on our master node. This allows us to keep the code up to date efficiently and effortlessly. At last, we’ve arrived at the final piece of the puzzle, a command to get the job going: spark-submit –master spark://<HOST:PORT> –packages <NECESSARY_PACKAGE_NAMES_SEPARATED_BY_COLON> –py-files <PATH_TO_THE_PYTHON_LIBRARIES> <PATH_TO_THE_PROGRAM> Port for the cluster master is 7077 by default. Also, –py-files argument is here to allow us to distribute required Python libraries and dependencies.In order to prevent the unwanted termination of your job in case SSH connection suddenly closes, add & at the end of the command. Our use case If you don’t follow the thread that our project is, here’s a little recap of what had been done until Spark came into play. We handled preliminary data analysis, datasource preparation, and architecture planning. Also, we set up Kafka in order to store our messages into different topics and partitions. Now the time has come to take the raw data stored inside Kafka and turn it into some useful insights. The sales data we’re dealing with could be transformed into various key performance indicators, famously abbreviated as KPIs, that could help us gain a much better understanding of what is really going on in the world of retail. We are storing them in Kafka because that will allow us to easily implement real-time notifications later in the project. Apart from