An Introduction to Apache Spark

MapReduce

MapReduce defines a programming pattern, where the data records get transformed into something else and then these results are reduced again. A typical "Hello World" application in MapReduce is a distributed word count. You have several files of text and want to count (and sum) the amount of each word in all documents.

To count the words in a "classic" way you may come up with such an implementation (pseudo-code):

result = {}

for l in open(f).lines():
    for token in l.split(" "):
        if token in result:
            result[token] += 1
        else:
            result[token] = 1
return result

This code is slow as the processing of the files is way faster than the reading of the file. Moreover, these counts can be done in parallel and can then be summed to reduce the waiting on the reading.

Lets define a map_line function, that gets as input a line of text. This will handle the splitting of the line and perform a word count on a line basis.

The next function will be a reduce_lines function, that take a list of mapped lines and reduces them to a dict again.

You can run the final code on your own if you want to play with it.

from typing import List

def map_line(line: str) -> dict:
    result = {}
    for token in line.split(" "):
        if token in result:
            result[token] += 1
        else:
            result[token] = 1

    return result

def reduce_lines(lines: List[dict]) -> dict:
    total_result = {}
    for line in lines:
        for word, count in line.items():
            if word in total_result:
                total_result[word] += count
            else:
                total_result[word] = count

    return total_result

data = ["test line we want to count", "and this line I want to count as well"]
mapped = map(lambda sentence: map_line(sentence), data)
reduced = reduce_lines(mapped)
print(reduced) # {'test': 1, 'line': 2, 'we': 1, 'want': 2, 'to': 2, 'count': 2, 'and': 1, 'this': 1, 'I': 1, 'as': 1, 'well': 1}

This separation of functions allows all maps to be run in parallel (what we don't do in this example). This is possible as the map is a pure function. A pure function don't share any state from outside of the function, which means it has no side effects. Running this code in parallel will result in a longer waiting for the underlying hard drive.

Instead of trying to improve the performance on one server, I will now introduce the Hadoop Distributed File System (HDFS).

HDFS

To allow parallel processing without the IO bound of one server, the data is distributed on several servers. When using an appropriate data format like csv, json (each object in one line), parquet or avro - this data can be read in parallel by an offset on several servers. Each server will only read a chunk of the entire file, allowing a high IO performance.

The downside is, that the MapReduce API is not intuitive and after each transformation, you have to store the data again. The next MapReduce job will read the data again and continue - you will "soon" get IO bound again.

Google removed MapReduce from their stack "recently" and replaced it by something else (we can guess, that's Apache Spark). I would consider MapReduce as deprecated technology but it's still good to know to understand Apache Spark better.

YARN

YARN stands for "Yet another resource negotiator" - this part of Hadoop will tell the jobs what server they can run on. YARN is also responsible for terminating applications that consume more memory than we allowed them to consume. We need to define memory limits as we can have a mixed setup of servers with a lot a RAM and with less memory - YARN will put the application where it can run and enough memory is available based on the memory settings we provide.

Pick a Cluster

Do you want to setup your own Hadoop? Sure you can to that but its not a trivial task.

• Buy Hardware (estimate at least 3 times the size of files you want to store for replications and backup)
• Setup Network
• generate ssh's keys that all severs can reach each other
• install java
• Install required components of the hadoop stack
• deploy all the settings to the clients

Do you really want to know each detail of your cluster and have the appropriate hardware and how-to available? For a local installation you way use a distribution that help with the details.

How much hardware do you need? This depends on the amount of data you are dealing with and how long your jobs are running.

As rule of thumb, it would at least go for a ration of 1 CPU to 8 GB of ram for each server in the cluster, that is doing computing tasks. Most of the jobs I am writing on a huge production cluster at least need 8 GB of ram. Some jobs even need 32GB of ram. When you need more, your code have a high chance to have space for improvement / reducing the memory usage.

In case you don't have a long running pipeline, you can try out managed solutions.

Managed Cluster

If you don't have a Hadoop cluster, you can put your data on common cloud storage like S3 from AWS to improve your processing speed or search for a managed solution. You can directly read from s3 within Apache Spark and other cloud vendors should work as well but I haven't tried.

Still you don't know what Apache Spark is and why to use it - let's fix it.

Caching

Apache Spark will cache the data you are reading from your source. In case you do something more with the data, you don't need to read it again. But you can also use caching explicitly, to help Apache Spark to improve the speed of your code even more. Say you are doing a heavy computation and then want to output the data in different data formats - instead of doing the computation twice, you can tell Apache Spark to cache the result just before you want to write it down.

At first the data will be processed normally but the second write operation will just use the cached data that was processed already and only apply minimal transformations required to save the data in a different format. Smart caching can also reduce the used memory of your jobs, when only one copy of the data have to be stored in memory instead of several "copies" of it - for a cross join for example.

Lazy Evaluation / DAG

Apache Spark will create a directed, acyclic graph of the processing steps down to any write operations. This graph gets optimized by Apache Spark and then executed. For example, when you first read all files from a directory but then you are really just interested in data from a particular date, all the other files can be skipped by Apache Spark. This will require some processing steps again (for the data storage) to allow Apache Spark to do such filtering as early as possible and get all optimizations. Lets now have a high level overview about Apache Spark components:

Apache Spark History Server

This is a webserver that allows to view the current steps that Apache Spark is executing and give you a lot details what is happening.

Apache Spark Driver

Your code gets first copied from a local system to YARN. YARN will check if resources are available (that you defined when you submitted the application) and send your application to the servers that will run your code. The driver does all low level coordination of the tasks with the executors. If you are interested on more details, checkout the links at the bottom of this post.

Apache Spark Executor

These servers do the heavy lifting and consume the memory of your job. In general, you have only one driver and a lot of executors. The more you can get running, the better in general as long as you don't hit memory issues.

Memory Management

When not enough resources are available for an application, then it needs to wait (YARN put the application in the ACCEPTED state).

Graphite

You still have no idea why your job is crashing? You need more insides.

Note that the following metrics are only written when run in cluster mode / on YARN.

Emetriq use graphite to log the production jobs. The integration to graphite is build into Apache Spark. Pass the following arguments to your spark-submit command, to write several (internal) Apache Spark metrics to graphite that you can use for further investigations:

--conf "spark.metrics.conf.*.sink.graphite.class=org.apache.spark.metrics.sink.GraphiteSink"
--conf "spark.metrics.conf.*.sink.graphite.host=graphiteHost"
--conf "spark.metrics.conf.*.sink.graphite.port=2003"
--conf "spark.metrics.conf.*.sink.graphite.period=10"
--conf "spark.metrics.conf.*.sink.graphite.unit=seconds"
--conf "spark.metrics.conf.*.sink.graphite.prefix=prod.<app>"
--conf "spark.metrics.conf.*.source.jvm.class=org.apache.spark.metrics.source.JvmSource"

Source: https://github.com/LucaCanali/Miscellaneous/blob/master/Spark_Dashboard/Spark_metrics_config_options.md