Introduction
Spark map() and mapPartitions() transformations apply functions to each element/record/row of the DataFrame/Dataset and return the new DataFrame/Dataset. In this article, I will explain the difference between map() vs mapPartitions() transformations, their syntax, and usages with Java examples.
- map() – Spark
map()transformation applies a function to each row in a DataFrame/Dataset and returns the newly transformedDataset. - mapPartitions() – This is exactly the same as map(); the difference being, Spark
mapPartitions()provides a facility to do heavy initializations (for example Database connection) once for each partition instead of doing it on every DataFrame row. This helps the performance of the job when dealing with heavy-weighted initialization on larger datasets.
Key Points:
- One key point to remember, these both transformations return the
Dataset<U>but not theDataFrame(In Spark 2.0,DataFrame = Dataset<Row>). - After applying the transformation function on each row of the input DataFrame/Dataset, these return the same number of rows as input but the schema or number of the columns of the result could be different.
- If you know
flatMap()transformation, this is the key difference between map and flatMap where map returns only one row/element for every input, while flatMap() can return a list of rows/elements.
Understanding StructType and StructField
Before diving into the examples, it's essential to understand StructType and StructField. StructType is a collection of StructFields that fully defines the schema of a DataFrame. Each StructField has a name, data type, and a flag indicating whether null values are allowed.
Let’s see the differences with an example. First, let’s create a Spark DataFrame:
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.*;
import org.apache.spark.sql.Row;
import java.util.Arrays;
import java.util.List;
List<Row> structureData = Arrays.asList(
RowFactory.create("James", "", "Smith", "36636", "NewYork", 3100),
RowFactory.create("Michael", "Rose", "", "40288", "California", 4300),
RowFactory.create("Robert", "", "Williams", "42114", "Florida", 1400),
RowFactory.create("Maria", "Anne", "Jones", "39192", "Florida", 5500),
RowFactory.create("Jen", "Mary", "Brown", "34561", "NewYork", 3000)
);
StructType structureSchema = new StructType(new StructField[]{
new StructField("firstname", DataTypes.StringType, true, Metadata.empty()),
new StructField("middlename", DataTypes.StringType, true, Metadata.empty()),
new StructField("lastname", DataTypes.StringType, true, Metadata.empty()),
new StructField("id", DataTypes.StringType, true, Metadata.empty()),
new StructField("location", DataTypes.StringType, true, Metadata.empty()),
new StructField("salary", DataTypes.IntegerType, true, Metadata.empty())
});
SparkSession spark = SparkSession.builder().appName("example").master("local").getOrCreate();
Dataset<Row> df2 = spark.createDataFrame(structureData, structureSchema);
df2.printSchema();
df2.show(false);
Output:
root
|-- firstname: string (nullable = true)
|-- middlename: string (nullable = true)
|-- lastname: string (nullable = true)
|-- id: string (nullable = true)
|-- location: string (nullable = true)
|-- salary: integer (nullable = true)
+---------+----------+--------+-----+----------+------+
|firstname|middlename|lastname|id |location |salary|
+---------+----------+--------+-----+----------+------+
|James | |Smith |36636|NewYork |3100 |
|Michael |Rose | |40288|California|4300 |
|Robert | |Williams|42114|Florida |1400 |
|Maria |Anne |Jones |39192|Florida |5500 |
|Jen |Mary |Brown |34561|NewYork |3000 |
+---------+----------+--------+-----+----------+------+
In order to explain map() and mapPartitions() with an example, let’s also create a “Util” class with a method combine(), this is a simple method that takes three string arguments and combines them with a comma delimiter. In real-time, this could be a third-party class that does complex transformation.
public class Util implements java.io.Serializable {
public String combine(String fname, String mname, String lname) {
return fname + "," + mname + "," + lname;
}
}
We will create an object for this class by initializing and calling the combine() method for each row in a DataFrame.
Spark map() transformation
Spark map() transformation applies a function to each row in a DataFrame/Dataset and returns the new transformed Dataset. As mentioned earlier, map() returns one row for every row in an input DataFrame, in other words, input and the result exactly contain the same number of rows.
For example, if you have 100 rows in a DataFrame, after applying the function map() returns with exactly 100 rows. However, the structure or schema of the result could be different.
Spark provides a map transformation that can be applied using the MapFunction in Java, and this transformation returns a Dataset<U> but not a DataFrame (which is Dataset<Row>). If you want a DataFrame as output, then you need to convert the Dataset to DataFrame using toDF().
Usage:
Dataset<Row> df3 = df2.map((MapFunction<Row, Row>) row -> {
// This initialization happens for every record
// If it is heavy initializations like Database connects
// It degrades the performance
Util util = new Util();
String fullName = util.combine(row.getString(0), row.getString(1), row.getString(2));
return RowFactory.create(fullName, row.getString(3), row.getInt(5));
}, RowEncoder.apply(new StructType()
.add("fullName", DataTypes.StringType)
.add("id", DataTypes.StringType)
.add("salary", DataTypes.IntegerType)));
df3.printSchema();
df3.show(false);
Since map transformations execute on worker nodes, we have initialized and created an object of the Util class inside the map() function, and the initialization happens for every row in a DataFrame. This causes performance issues when you have heavily weighted initialization.
Note: When running on Standalone mode, initializing the class outside of the map() still works as both executors and driver run on the same JVM but running this on a cluster fails with exception.
The above example yields the following output.
root
|-- fullName: string (nullable = true)
|-- id: string (nullable = true)
|-- salary: integer (nullable = false)
+----------------+-----+------+
|fullName |id |salary|
+----------------+-----+------+
|James,,Smith |36636|3100 |
|Michael,Rose, |40288|4300 |
|Robert,,Williams|42114|1400 |
|Maria,Anne,Jones|39192|5500 |
|Jen,Mary,Brown |34561|3000 |
+----------------+-----+------+
As you notice the above output, the input DataFrame has 5 rows, so the result of the map also has 5, but the column counts are different.
Spark mapPartitions() transformation
Spark mapPartitions() provides a facility to do heavy initializations (for example Database connections) once for each partition instead of doing it on every DataFrame row. This helps the performance of the job when you are dealing with heavy-weighted initialization on larger datasets.
Syntax:
<T> Dataset<U> mapPartitions(MapPartitionsFunction<T,U> func, Encoder<U> encoder);
mapPartitions keeps the result of the partition in-memory until it finishes executing all rows in a partition.
Usage:
Dataset<Row> df4 = df2.mapPartitions(iterator -> {
// Do the heavy initialization here
// Like database connections, instantiation etc.
Util util = new Util();
List<Row> res = new ArrayList<>();
while (iterator.hasNext()) {
Row row = iterator.next();
String fullName = util.combine(row.getString(0), row.getString(1), row.getString(2));
res.add(RowFactory.create(fullName, row.getString(3), row.getInt(5)));
}
return res.iterator();
}, RowEncoder.apply(new StructType()
.add("fullName", DataTypes.StringType)
.add("id", DataTypes.StringType)
.add("salary", DataTypes.IntegerType)));
df4.printSchema();
df4.show(false);
This yields the same output as above.
SUMMARY: Converted the lesson content from Scala Spark to Java Spark using Java-specific classes and methods.