Spark for Hadoop MR Refugees

Spark computational model can be confusing for someone coming from the Hadoop map reduce paradigm. Don’t get me wrong, I think that previous experience with Hadoop is definitely very good to understand a spark job execution and at the end at a very fundamental simplified level , spark DAG pipeline can be reduced to operations executed transforming data local to the node (map) and operations that trigger a shuffle of the data to be processed in other node (reduce). But I think that at the beginning you will find yourself wondering, I did this this or that way with Hadoop, how it is done with Spark?.

Distributed execution and lazy model

When writing a map reduce job, it is clear where each piece is executed and when. You have the driver program that is executed in the client/frontier node. Then you have mapper, reducer, combiner task classes that are distributed to the cluster and run in parallel processing the data. Those are the distributed code parts , that will be serialized, cannot communicate with the other tasks, etc…. You know that the distributed execution will be started with the method waitForCompletion on the driver, and the result will be written to a file using a configured outputformat and then the job ends.

With spark the code in the driver is more “interactive”, the driver can trigger distributed tasks, get some results, launch more distributed tasks and so on. Is a bit more difficult to identify when the code will be executed and when.

The  key here is that the inputs and outputs for the distributed tasks in spark are the RDD objects that abstract a distributed dataset managed by the framework that can be stored in the cluster either on disk or memory. That means that the code that will be distributed to the cluster will be the objects passed to the RDD methods. So all RDD operations are distributed, the parameter objects will be serialized and sent over the wire and you must take in account the closure etc...

The map reduce tasks that we know from Hadoop can be translated in spark as map/flatmap and reduceByKey. This is using key/value RDDs (Hadoop operates always with key/value pairs, spark support RDDs of simple types for example). Now, when does the driver executes these operations?. Spark uses a lazy evaluation model meaning that for certain operations called transformations the execution will be delayed till an actual result is demanded. These map / reduceByKey operations are transformations, so if you chain a set of map reduce jobs using them, they will be added to an internal execution graph pipeline (DAG) and delayed till a result is requested with another operation called action. When the execution is triggered the current graph is optimized/reordered and the tasks executed in the cluster.

In Hadoop these action operations correspond to the waitForCompletion call that usually will involve the output written to a file. There is an action for that in Spark for example saveAsTextFile. But you can request results from RDD transformations without explicitly writing to file using actions like collect or take. Moreover, once and RDD is transformed after an action you can issue new transformations calls to this RDD until a new action is triggered. With Hadoop that can only be done chaining several independent jobs.

Basic Map Reduce: Mapper/reducer/combiner equivalents

     These basic Hadoop tasks are now RDD transformations in Spark.

     The mapper task equivalent in spark using the map/flatmap transformation on an k/v pair RDD. Use map when we want to emit one output per input and flatmap when we want several outputs per input, flattening them afterwards.

     The reducer task equivalent in spark using the reduceByKey/groupByKey transformation on an k/v pair RDD. reduceByKey does not allow us to change the key and assumes that the reducer function is associative executing a previous combiner step before shuffling. groupByKey is more flexible in the sense that it allows to change the key it does not apply any reduce function and only groups the values by key, so there are potential risks of heavy I/O and out of memory.

The classic word count example:

val wordCounts = inputData.flatMap(line => line.split("\\s* \\s*"))

.filter(word => word.size>0)

.map(word => (word.toLowerCase,1))

.reduceByKey(_+_)

   
As we know with Hadoop there are cases when we cannot use the same reducer function as the combiner because the input and output does not match. By default the reduceByKey transformation executes the reduce function locally in the node as a combiner before shuffling and applying it again as a reducer. If we want to use a different function as combiner we need the combineByKey transformation where we can specify the reducer and combiner functions separately.

An example of different combiner and reducer, we are getting the words average length grouped by the first character:

val wordAvgLengthByFirstChars = firstCharAndCounts.combineByKey((_,1),

 (x:(Int,Int),y) => (x._1+y,x._2+1),

 (x:(Int,Int),y:(Int,Int)) => (x._1+y._1,x._2+y._2))

.mapValues(x => (x._1/x._2).toFloat)

Other concepts: Partitioner, GroupComparator, setup, cleanup

     In Hadoop and spark the default partitioner is based on the hash code of the key, as you know you can define a custom partitioner in Hadoop setting the custom partitioner class in the configuration of the job. In spark, guess what, there is an RDD transformation to define a custom partitioner call partitionBy. 

     Group comparators were used with Hadoop when implementing custom composite keys, like in the secondary sort pattern, and control the pairs will be grouped in the reducer phase. The same can be achieved with Spark using groupBy transformation with a custom comparator, as in the groupByKey transformation , use with case since all the values belonging to the same key must fit in memory

     Setup and clean up can be used in the mapper stage with the mapPartitions that run once for each partition and traverses an iterator of all the values, So you can execute your setp code before starting to iterate and the clean up one once the call to iterator has next returns false.

     For the reducer case you have to use the groupByKey transformation previously and then mapPartitions.

Some Map reduce patterns

 

I’m not going into much detail about what these basic Map Reduce patterns are, just giving an idea of how could they be implemented with Spark,

- In mapper combining.

     The reasons for this Hadoop mr pattern were mainly two: perform the combining step in memory and assure it was executed, since Hadoop did not guarantee the execution of the combining step and it was performed on the output file of the mapper. This is not the case with spark but still there could be an scenario for this pattern, when we want to use the groupByKey transformation, but we want to minimize the data shuffled performing a local combining in each partition.  Don't forget that the local aggregated data must fit in memory.

A word count example using in mapper combining and groupByKey:

val inMapperCombWordCount = inputData.mapPartitions { pIterator => val localComb = Map[String,Int]();

pIterator.foreach{_.split("\\s* \\s*")

  .map{x=>x.toLowerCase;

  localComb(x) = localComb.get(x).getOrElse(0)+1}};

localComb.toIterator}

.groupByKey.mapValues(_.sum)

- Secondary sort

     The elements needed for this pattern are the custom partitioner for the composite key and the custom group comparator. We can achieve that using the groupBy method with a custom partitioner and a custom comparator. This method is expensive in terms of the amount of data shuffled.

- Reducer side joins

     This pattern is already included in spark as the join transformation. In Hadoop this involves shuffling all the data belonging to the datasets being joined so the data belonging to the same key goes to the same partitioner. Spark adds a very important optimization, it keeps track if one of the datasets has been already partitioned and only shuffles the other, if both have been partitioned with the same partitioner and are cached in the same machines then no shuffling at all will occur. 

- Optimized Serialization

One of the key points in Hadoop is the provided optimized serialization system using writable objects that can be easily extended. Spark relies on a third party library, kryo.  

Pangool, The Hadoop Companion

Writing mapreduce jobs with Hadoop is no trivial task, the programming model has a lot of moving parts that have to work together: mapper, combiner, reducer and in most cases partitioner, comparators, writables and writablecomparable objects.

As you know, there are tools that provide a higher level of abstraction allowing us to define the jobs by using a declarative (Hive) or procedural (Pig) notation and then generating and launching the java mapreduce tasks transparently.

This is often enough in many cases, for example when we are executing analytical or exploratory queries against a data set, the performance of the map reduce job is not a priority, it is more important to be able to create these queries easily and with a familiar, sql-like language. Not everybody is a Java expert, after all.

In other cases performance can be a key requirement and you need to roll up your sleeves and write the java code directly. This is particularly important in complex processes that involve several chained jobs. If you can reduce the number of jobs, the optimization is worth the effort.

It would be nice to have a tool that help us to abstract from the more frequent and tedious mapreduce patterns without losing the flexibility and tuning capabilities of the Hadoop java api. Enter Pangool, a Java low level mapreduce api built on top of Hadoop. By implementing an intermediate Tuple-based schema it allow us to implement patterns like secondary sorting or reduce-side joins almost transparently in a declarative fashion. This tuple schema free us of the hassle of implementing custom writable objects and is leveraged via the provided tuple input and output formats to seamlessly build job pipelines. All of this is achieved without losing the flexibility and performance that Hadoop low level api provides.

To see an example of a job created with Pangool we can rewrite the job from the previous post. We will see how it simplifies the application of the secondary sort pattern and how we can still implement our own customizations (custom partitioner, input file format...)

The code is available at github

The first change we notice is the definition of tuple schemas:

1 2 3 4 5 6 7 8

private static final Schema INTERMEDIATE_SCHEMA = new Schema( "schema", Arrays.asList(new Field[] { Field.create(SPLIT_FIELD, Type.INT), Field.create(LINE_FIELD, Type.STRING, true), Field.create(SEQ_FIELD, Type.LONG), Field.create(TOTAL_FIELD, Type.LONG, true), })); private static final Schema RESULT_SCHEMA = Mutator.subSetOf("mutated", INTERMEDIATE_SCHEMA, SEQ_FIELD, LINE_FIELD);

With these two lines we define the output of the mapper process (intermediate schema) and the output of the job (result schema). That's all, we don't need to create custom writables for the composite keys and values used in the secondary sort pattern  and we can support null values in a field to send the split total count with a line value of null  (the boolean set to true in the field definition indicates null support).

Note that we can easily reuse our original schema to define the output using the Mutator class.

Now we want to apply the secondary sort, ordering by sequence and grouping by split:

1 2 3

mr.addIntermediateSchema(INTERMEDIATE_SCHEMA); mr.setGroupByFields(SPLIT_FIELD); mr.setOrderBy(new OrderBy().add(SPLIT_FIELD, Order.ASC).add(SEQ_FIELD, Order.ASC));

We indicate the schema, group field, and the ordering, and it's finished, this is our implementation of the secondary sort pattern (This is in addition to the mapper and reducer of course) . Easy, don't you think?. And all of this without losing the flexibility of Hadoop low level api, we still provide our custom input format, and input split classes:

1	mr.addInput(new Path(input), new SequencerInputFormat(), new SequencerMap());

and the custom partitioner:

1 2	Job job = mr.createJob(); job.setPartitionerClass(SequencerPartitioner.class);

To retrieve the number of splits we have to use the classic api:

1 2 3

FileInputFormat.addInputPath(job, new Path(input)); job.getConfiguration().setInt(TOTAL_SPLIT_NUMBER, new TextInputFormat().getSplits(job).size());

There are some small additional code changes derived from the new api, for example to recover our custom split instance in the mapper:

1 2	SequencerFileSplit split = (SequencerFileSplit) ((TaggedInputSplit) context .getHadoopContext().getInputSplit()).getInputSplit();

But in overall we are still using the mapreduce model, just with schema defined tuples instead of key value pairs and we can still leverage the underlying framework,

I have used Pangool in several projects, and in my opinion it provides the right balance between abstraction and low level capabilities. The advantages are more evident when you need to chain a sequence of jobs and perform joins, that could be a subject for a future post. The paradigm shift to the tuple model is small since it is an evolution from the previous api, you still will take advantage of all your hard learned mapreduce skills.

Hadoop: Leveraging the framework

Map-Reduce programming model has become very popular thanks to Hadoop. It provides a simple way to parallelize processing through the distribution of the computation to the nodes where the data is located. You are free from the hassle of traditional multi-threaded distributed programming (locks, signaling, etc..), since each process runs independently in his own virtual machine. This has also disadvantages, mainly the lack of communication between processes, Mappers and Reducers are completely oblivious of the status of the rest of processes. As a truly functional paradigm, the only input the Reducer process have are the intermediate output files from the map stage.

This can prove a difficulty for the parallelization of some algorithms or even render it infeasible. In other cases the limitations can come from the framework itself. For example, some information on the input must be shared by all the tasks and is not immediately available. With Hadoop this is can be often overcome using internal mechanisms or extending the framework, although not very straightforwardly sometimes.

Let's elaborate with an actual case:

We have a huge input text file in HDFS and we want to process it adding unique sequence ids to each line, This ids will start with 0 and will be sequentially increasing in steps of one, so for a file of N lines we will add to each line an id starting in 0 and ending in N-1. The output will be the same file but having each line prepended with the id.

Our requirements are:

- Input will be a text file.

- As said the sequential ids will start in 0 and end in N-1 (N being the total number of lines of the file) in increments on 1

- The output will be in the same order of the input, we don't want to shuffle lines.

- Only one map-reduce job will be used.

- Last and not least, the whole point of using Hadoop is to parallelize and scale so no one-reducer jobs.

Summarizing, we are numbering the lines of a file in a distributed fashion using a Hadoop job.

The Job

 

- At the mapper phase, get an id of the file split currently processed, we need this id to be unique and sequentially ordered starting with the first block of the file. Then iterate this split lines incrementing a counter and emitting as key a pair of (splitId, counter). and as value the pair (counter, line) . As you can see we repeat one field of the composite key on the composite value, we will need that for implementing a secondary sorting pattern. The mapper will group all the lines with the same splitId adding the sequence information we can later retrieve in the reducer to maintain the same ordering.

- In the mapper cleanup process we need to propagate the total number of lines of the processed split to all the reducers emitting a pair with key (splitId, currentsplit) and value (currentsplit, total count) for every split. We only emit the to the splits Id greater than the current one.  At the reducer stage we will need this previously calculated totals to set the initial sequence number.

 If we take the following splits:  
 Split 1:  
 A  
 B  
 C  
 Split2:  
 D   
 E  
 Mapper 1 emits: ((1,1) (1,A))  ((1,2) (2,B))  ((1,3) (3,C))  
 Then in the clean-up we inform the reducers processing the split 2 that the total count for split 1 is 3:  
 ((2,1)(1,3))  
 Similarly Mapper 2 emits:  
 ((2,1) (1,D))  ((2,2) (2,E))  
 In the clean-up process we don't need to emit anything, since this is the last split (2 of 2).  

-We apply a secondary sort pattern, grouping in the reducers for split id and receiving the pairs ordered by the counter sequence, we need also to identify pairs with the mapper calculated totals and direct them to each partition, more on that later:

- For each key the reducer checks the count totals coming from the splits with an id lesser than the current key id, sums them and begin writing to the output file the lines using that total as the start value and incrementing with steps of 1. If there are no splits with a lower id, we begin with 0 as the first sequence id.

 In the above example the reducers will receive two calls, (remember we group by split id and we have two splits):  
 (1, (1,A) (2,B) (3,C))  
 (2, (1,3) (1,D) (2,E))  
 For split 1 we have no previous split total so we start with 0 and write to the file:  
 (0, A)(1,B)(2,C)  
 for the next split 2, we have one key (1) lower than the current with a total of 3: (1,3), so we start on 3:  
 (3,D) (4,E)  

 

 Java code

You can find the source code in my github repository, There are three classes: two writable implementations for the composite key and value and the job implementation class including all the moving parts (mapper, reducer, partitioner...): SplitSequenceWritable.java, SequenceLineWritable.java, SequencerJob.java

The key points here are:

- We need to emit an unique sequential split Id in the mapper and we need also to know the total number of splits. 

 

We can do that extending the framework. The list of FileSplits for the input file is calculated in the getSplits method of the TextInputFormatClass using as parameter the job context, so we can extend that class and also the FileSplit class to create instances with the total count and the split sequence id information:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35

public static class SequencerFileSplit extends FileSplit { private int totalSplits; private int splitId; SequencerFileSplit() { super(); } SequencerFileSplit(FileSplit split, int splitId, int totalSplits) throws IOException { super(split.getPath(), split.getStart(), split.getLength(), split.getLocations()); this.splitId = splitId; this.totalSplits = totalSplits; } @Override public void readFields(DataInput in) throws IOException { super.readFields(in); splitId = in.readInt(); totalSplits = in.readInt(); } @Override public void write(DataOutput out) throws IOException { super.write(out); out.writeInt(splitId); out.writeInt(totalSplits); } //Rest of methods }

We add our custom fields to the split, note that this is a writable object distributed in the cluster so you need to make sure the new fields are serialized and also provide an empty constructor.

Our implementation of getSplits simply calls to the parent method to retrieve all the splits information and returns them wrapped in out custom split with the id and total information. We are assuming here that the framework returns the splits ordered by start offset and that is actually true:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

public static class SequencerInputFormat extends TextInputFormat { @Override public List<InputSplit> getSplits(JobContext arg0) throws IOException { List<InputSplit> splits = super.getSplits(arg0); int total = splits.size(); List<InputSplit> result = new ArrayList<InputSplit>(total); int counter = 1; for (InputSplit split : splits) { result.add(new SequencerFileSplit((FileSplit) split, counter++, total)); } return result; } }

- We need to group the lines by splitIds with the same ordering of the input file. The output must maintain the initial order.

 

We are implementing a canonical secondary sort pattern with custom writable composite keys and values, a partitioner and a custom GroupComparator. The composite key containing splitid and sequence number means that we will group by splitId and order by the input sequence. We need to implement a groupcomparator by splitId only and a partitioner to send the pairs to the reducers based also on the splitId. To maintain the input file ordering the partitioner must send the adjacent splits to the same reducer. For example imagine we have 12 splits and 3 reducers, we want splits 1,2,3,4 to go to one reducer, 5,6,7,8 to another and 9,10,11,12 to the last one. To do that we need to know the total split count value in the partitioner. 

This class is not aware of the job configuration by default, but we can make Hadoop to inject it automatically implementing the Configurable interface:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

public static class SequencerPartitioner extends Partitioner<SplitSequenceWritable, SequenceLineWritable> implements Configurable { private Configuration conf; private int totalSplitNumber; @Override public int getPartition(SplitSequenceWritable key, SequenceLineWritable value, int numPartitions) { return ((key.getSplit() - 1) * numPartitions) / totalSplitNumber; } @Override public Configuration getConf() { return conf; } @Override public void setConf(Configuration conf) { this.conf = conf; totalSplitNumber = conf.getInt(TOTAL_SPLIT_NUMBER, 0); } }

And now we can make it available in the job configuration, creating an instance of TextInputFormat to retrieve the value:

1	job.getConfiguration().setInt(TOTAL_SPLIT_NUMBER, new TextInputFormat().getSplits(job).size());

The reducer calculates the start number for out split id keys, remember we emitted that totals in the mapper as "meta-pairs" for all the splits:

 (split1, -currentsplit) (-1, currentSplitLineCount) 

 (split2, -currentsplit) (-1, currentSplitLineCount)

 (split3, -currentsplit) (-1, currentSplitLineCount)

.

.

As you see the sequence value corresponding to the total was emitted as -1 in the mapper stage. This is an easy way to identify an aggregation pair since all sequences are positives. Additionally the reducer will receive these totals as the first pairs since the iterable is ordered. Once we have the start sequence number we can iterate the rest of pairs (also coming in the right order thanks to the secondary sorting) containing the lines and number them incrementally:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

protected void reduce(SplitSequenceWritable key, Iterable<SequenceLineWritable> values, Context context) throws IOException, InterruptedException { Iterator<SequenceLineWritable> iterator = values.iterator(); long sequenceNumber = 0; SequenceLineWritable value = iterator.next(); while (iterator.hasNext() && value.getSequence() < 0) { sequenceNumber += Long.valueOf(value.getLine()); value = iterator.next(); } sequence.set(sequenceNumber); line.set(value.getLine()); context.write(sequence, line); sequenceNumber++; while (iterator.hasNext()) { value = iterator.next(); sequence.set(sequenceNumber); line.set(value.getLine()); context.write(sequence, line); sequenceNumber++; } }

We have extended the framework to be able to identify the current split in each stage (map, partitioner, comparator, reducer). This information needs to be calculated by Hadoop anyway and by making it available we gain insight and flexibility in our jobs. Since the pairs are grouped by split in the reducers, the data is uniformly distributed to avoid hot spots nodes allowing for a linear scaling. More generally this pattern could be used to apply transformations to the lines of a text file maintaining the ordering.