Partitioning

RDDs are divided into partitions which reside on different nodes. A single partition never resides on more than one node. Each node must have at least one partition - otherwise it can not participate in RDD processing. By default, the number of partitions of a data set equals the number of processing units in the cluster. So if we have 4 nodes with 2-core CPU, by default we’ll end up with 8 partitions. These defaults can be changed.

The partitioning policies of Spark can be customised only for pair RDDs, because the custom partitioning policies are based on the pair keys.

Spark supports two customisable policies - Hash and Range partitioning. Hash partitioning distributes the data uniformly across all partitions based on the keys’ hash code. More formally, a pair <k,v> will land on partition p if p = k.hashCode % numPartitions.

While hash partitioning is applicable for any pair RDD, range partitioning implies that keys can be ordered and compared. For example, an RDD whose keys are integers or dates could be partitioned by range. Spark would sort the data and then split it into equal size partitions so that pairs whose keys are “close” are located together.

To apply a custom partitioning, we can use the partitionBy method of RDD. It takes as an argument an instance of Partitioner, which could be either RangePartitioner or HashPartitioner. After an RDD is partitioned, it’s imperative to persist it, which will effectively distribute the data across the nodes. If we fail to persist, the reshuffling will be re-evaluated lazily again and again.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// Let's create an Rdd and a pair RDD of odd and even numbers:
val rdd = sc.parallelize((1 to 100).toList);
val pairRdd = rdd.map(i => (i,i));

// Create a custom partitioner with 10 partitions.
// We also need pass the target pairs, so the partitioner
// can compute the range.
val ranger = new RangePartitioner(10, pairRdd);

// Create a partitioned version of the RDD
// We need to persist to avoid re-evaluation!
val rangedPairRdd = pairRdd.partitionBy(ranger).persist();

// Create a custom hash partitioner with 100 partitions.
val hasher = new HashPartitioner(10, pairRdd);

// Create a partitioner version of the RDD
// We need to persist to avoid re-evaluation!
val hashedPairRdd = pairRdd.partitionBy(hasher).persist();

RDDs are often constructed by transforming existing RDDs. In most cases, the resulting RDD uses the partitioner of the source RDD. However, in some cases a new partitioner is created for the child. For example, the sortByKey transformation always results in an RDD which uses a RangePartitioner. Alternatively, groupByKey always uses a HashPartitioner.

There are also transformations like map and flatMap which produce RDDs without partitions. This is because they can change the key based on which the partition has be defined. In fact, any transformation that can change the pairs’ keys results in partition loss. Hence, functions like mapValues should be preferred to their alternative like map.

1
2
3
4
5
6
7
val rdd1: Rdd[(Int, Object)] = // Load the RDD from somewhere
val rdd2: Rdd[(Int, Object)] = // Load the RDD from somewhere

val pred = // some predicate function

// Solution 1 - join and filter
val result = rdd1.cartesian(rdd2).filter({case (p1, p2) => p1._1 === p2._1 }).filter(pred);

1
2
// Solution 2 - join and filter
val result = rdd1.join(rdd2).filter(pred);

1
2
3
// Solution 3 - join and filter
val pred = // some predicate function
val result = rdd1.filter(pred).join(rdd2.filter(pred));

1
2
3
4
import org.apache.spark.sql.SparkSession;

// Create the session
val sparkSession = SparkSession.builder().appName("App").getOfCreate();

1
2
3
4
5
6
7
8
9
10
11
12
// Create a DataFrame by specifying the column names
val rdd = ... // e.g. RDD[(Int, String, String)]
val dfNames = rdd.toDF("id", "firstName", "lastName");

// Create a DataFrame without names - spark will name the columns as _1, _2 ..
val dfNoNames = rdd.toDF();

// Create a DataFrame from an RDD of a case class.
// Spark will automatically assign the field names to the columns:
case class Person(id: Int, firstName: String, lastName: String);
val rddPerson: RDD[Person] = // create and RDD[Person]
val dfPerson = rddPerson.toDf()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// Convert the RDD into RDD[Row]
// A Row is basically a sequence of strings and primitives
val someRdd: RDD[String] = ... // e.g. load a text file 
val rowRdd = someRdd.map(_.split(" ")).map(e => Row(e[0].toInt, e[1], e[2]);

// Define the schema of the individual fields:
val fields = Array(
  StructField("id", IntegerType, nullable = true),
  StructField("firstName", StringType, nullable = true),
  StructField("lastName", StringType, nullable = true)
)
val schema = StructType(fields);

// Create the DataFrame via the SparkSession instance
val df = sparkSession.createDataFrame(rowRdd, schema);

1
2
val dfJson = sparkSession.read.json("/some/file.json");
val dfCsv = sparkSession.read.csv("/some/file.csv");

1
2
3
4
5
6
7
8
9
10
11
12
// Create some data frame
val df = ... // e.g. load it from a CSV file

// Register the data frame as "people"
// This is the name we'll use from SQL
df.createOrReplaceTempView("people");

// Run the SQL with the SparkSession utility method
val johnsDf = sparkSession.sql("SELECT * FROM people WHERE firstName = \"john\"");

// We can also do it by invoking the corresponding methods of the frame
val johnsDf2 = df.select("firstName", "*").where("firstName", "john")

1
2
3
4
5
// Will print the first 20 rows and the headers
df.show();

// Will print debug information about the structure/schema
df.printSchema();

1
df.where($"id" > 100)

1
2
3
4
5
// Using where
df.where("id > 100");

// Using filter - can do more complex expressions. Note the use of "==="
df.filter(($"id" > 100) && ($"fistName" === "john"));

1
2
3
4
5
6
7
// Using plain SQL in a string
val result1 = sparkSession.sql("SELECT sum(id) FROM people GROUP BY firstName");

// Analogous to the above query using individual method.
// "groupBy" returns an instance of RelationalGroupedDataset
// "agg" gets us back in RDD land.
val result2 = df.groupBy($"firstName").agg(sum($"id"));

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// Removes rows that contain null or NaN in ANY column
var newDf = df.drop();

// Removes rows that contain null or NaN in ALL column
newDf = df.drop("all");

// Removes rows that contain null or NaN in the given columns
newDf = df.drop(List("id", "firstName"));

// Fills all missing numeric column values with 0
// and all missing string values with ""
newDf = df.fill(0).fill("");

// Fills all missing values for the specified columns
newDf = df.fill(Map("id" -> 0, firstName -> "john"));

// Replaces all firstName values of "john" with "peter"
newDf = df.replace(Array("firstName"), Map("john" -> "peter"));

1
2
3
4
5
6
// Inner join on the id columns
val innerJoin = df1.join(df2, df1.$"id" === df2.$"id");

// Rigth outer join on the id columns
// The supported join types are `outer`, `left_outer`, `leftsemi`:
val innerJoin = df1.join(df2, df1.$"id" === df2.$"id", "rigth_outer");

1
type DataFrame = Dataset[Row];

1
2
3
4
5
6
7
8
9
10
11
12
13
import spark.implicits._

// Create a Dataset from a DataFrame based on implicits
val ds = df.toDS

// Create a Dataset from an RDD based on implicits
val rddDs = rdd.toDS

// Create a Dataset from a collection on implicits
val collectionDs = List("a", "b").toDS

// Read a DataFrame from json, and convert to a Dataset of a type (e.g. Person)
val dsJson = sparkSession.read.json("/some/file.json").as[Person];