Working with the Spark Key/Value API

In this chapter, we'll be working with the Spark key/value API. We will start by looking at the available transformations on key/value pairs. We will then learn how to use the aggregateByKey method instead of the groupBy() method. Later, we'll be looking at actions on key/value pairs and looking at the available partitioners on key/value data. At the end of this chapter, we'll be implementing an advanced partitioner that will be able to partition our data by range.

In this chapter, we will be covering the following topics:

• Available actions on key/value pairs
• Using aggregateByKey instead of groupBy()
• Actions on key/value pairs
• Available partitioners on key/value data
• Implementing a custom partitioner

Available actions on key/value pairs

In this section, we will be covering the following topics:

• Available transformations on key/value pairs
• Using countByKey()
• Understanding the other methods

So, this is our well-known test in which we will be using transformations on key/value pairs.

First, we will create an array of user transactions for users A, B, A, B, and C for some amount, as per the following example:

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B", 10),
 UserTransaction("C", 10)
 )

We then need to key our data by a specific field, as per the following example:

val keyed = data.keyBy(_.userId)

We will key it by userId, by invoking the keyBy method with a userId parameter.

Now, our data is assigned to the keyed variable and its type...

Using aggregateByKey instead of groupBy()

In this section, we will explore the reason why we use aggregateByKey instead of groupBy.

We will cover the following topics:

• Why we should avoid the use of groupByKey
• What aggregateByKey gives us
• Implementing logic using aggregateByKey

First, we will create our array of user transactions, as shown in the following example:

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B", 10),
 UserTransaction("C", 10)
 )

We will then use parallelize to create an RDD, as we want our data to be key-wise. This is shown in the following example:

 val data = spark.parallelize(keysWithValuesList)
 val keyed = data.keyBy(_.userId)

In the preceding code, we invoked keyBy for userId to have the data of payers, key, and user...

Actions on key/value pairs

In this section, we'll be looking at the actions on key/value pairs.

We will cover the following topics:

• Examining actions on key/value pairs
• Using collect()
• Examining the output for the key/value RDD

In the first section of this chapter, we covered transformations that are available on key/value pairs. We saw that they are a bit different compared to RDDs. Also, for actions, it is slightly different in terms of result but not in the method name.

Therefore, we'll be using collect() and we'll be examining the output of our action on these key/value pairs.

First, we will create our transactions array and RDD according to userId, as shown in the following example:

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B"...

Available partitioners on key/value data

We know that partitioning and partitioners are the key components of Apache Spark. They influence how our data is partitioned, which means they influence where the data actually resides on which executors. If we have a good partitioner, then we will have good data locality, which will reduce shuffle. We know that shuffle is not desirable for processing, so reducing shuffle is crucial, and, therefore, choosing a proper partitioner is also crucial for our systems.

In this section, we will cover the following topics:

• Examining HashPartitioner
• Examining RangePartitioner
• Testing

We will first examine our HashPartitioner and RangePartitioner. We will then compare them and test the code using both the partitioners.

First we will create a UserTransaction array, as per the following example:

 val keysWithValuesList =
 Array(
 UserTransaction("...

Implementing a custom partitioner

In this section, we'll implement a custom partitioner and create a partitioner that takes a list of parses with ranges. If our key falls into a specific range, we will assign the partition number index of the list.

We will cover the following topics:

• Implementing a custom partitioner
• Implementing a range partitioner
• Testing our partitioner

We will implement the logic range partitioning based on our own range partitioning and then test our partitioner. Let's start with the black box test without looking at the implementation.

The first part of the code is similar to what we have used already, but this time we have keyBy amount of data, as shown in the following example:

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("...

Summary

In this chapter, we first saw available the transformations on key/value pairs. We then learned how to use aggregateByKey instead of groupBy. We also covered actions on key/value pairs. Later, we looked at available partitioners like rangePartitioner and HashPartition on key/value data. By the end of this chapter, we had implemented our custom partitioner, which was able to assign partitions, based on the end and start of the range for learning purposes.

In the next chapter, we will learn how to test our Spark jobs and Apache Spark jobs.