Gain insights into Spark Java API, learn about data transformations and SQL operations, and discover how to integrate big data and Java for scalable, high-speed processing.

tarball.tar.gz

DataFrameBasicsRun

SparkDataframeExampleDB

This course serves as a comprehensive introduction to the Spark Java API. Experienced Java developers will use object-oriented programming (OOP) principles to apply theory to Apache Spark and big data practice.

You’ll learn the basic components and architecture of Spark, a leading framework for building big data applications, before implementing them in Java. You’ll also explore data transformations like grouping, sorting, and joining. Further, you’ll learn to support SQL operations in the database and create a big data and batch template application with Java.

By the end of the course, you’ll be familiar with Apache Spark and know how to integrate big data with Java environments through the Spark Java API. You’ll wrap up by learning about monitoring and support functions for a live Spark Java environment.

Combining a leading big data framework with a leading programming language, this course will empower you to work efficiently with large volumes of data, and process at scale and speed.

Mastering Big Data with Apache Spark and Java

# FlatMap

The FlatMap operation is an old resident of the functional programming paradigm realm. It can be tricky to understand conceptually. There are two key components to learn about regarding FlatMap’s purposes:


1. Being a map transformation in nature, it applies a function to each element of a collection. This is no different than the plain `map()` function described before.

2. If the input is a collection of collections of elements (say a List of Lists, an array of arrays), it flattens the results into a single collection.


So fundamentally, objects are transformed in `map` and `flatMap` operations based on a function, but how the elements are processed differs. The former processes a single collection while the latter processes nested collections. 

In Spark, the concept is similar, but it displays some differences, so let’s start by visualizing this graphically and then practicing it in the code example.




Before diving into the code snippets, let’s draw an analogy of `flatMap` between Java API and Spark API.

In Java, `flatMap` flattens or expands into a collection of single elements. For instance, a list of all sub collections [{1,2} {3,4}, {5,6}] would be flattened to [1,2,3,4,5,6]. Moreover, if the map component of a function is applied, meaning if a number is added to each of the nested collections, then the result would be [2,3,4,5,6,7].

The same occurs in the homologous Spark transformation, but it also provides the option of applying this operation only to specific columns that hold collections within them. This is represented in the diagram by ‘Col2’ or the second column in each row of the DataFrame.

Furthermore, this operation is carried out row by row, until each Dataset’s row is flattened. So, a good way of looking at the rows then is to see them each as a collection of elements.
 
Let’s take a look closely at the situation explained in the previous subsection to clear any doubts we might have.

> **Note:** Project Interactivity Disclaimer: Because some transformations might produce a bit of extra output (i.e. Distinct) while working on the partitions of a DataFrame, this project includes break points in the code, so please be attentive at the output screen while running the application, as it requires user input to display all the operations in steps.





##############################
## Java
##############################
.mtj.tmp/
*.class
*.jar
*.war
*.ear
*.nar
hs_err_pid*

##############################
## Maven
##############################
target/
pom.xml.tag
pom.xml.releaseBackup
pom.xml.versionsBackup
pom.xml.next
pom.xml.bak
release.properties
dependency-reduced-pom.xml
buildNumber.properties
.mvn/timing.properties
.mvn/wrapper/maven-wrapper.jar

##############################
## IntelliJ
##############################
out/
.idea/
.idea_modules/
*.iml
*.ipr
*.iws
##############################
## MacOS Files
##############################
.DS_Store

# Spark Basic Transformations Part II Example Project

Project demonstrates usage of Basic Transformations (flatmap and distinct) in Spark Java Library.

To run execute:

`mvn install exec:exec`


<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.jmb.examples</groupId>
    <artifactId>basic-transformations-II</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <java.version>1.8</java.version>
        <scala.version>2.12</scala.version>
        <spark.version>3.0.0</spark.version>
        <maven-compiler-plugin.version>3.8.0</maven-compiler-plugin.version>
        <derby.version>10.14.2.0</derby.version>
    </properties>

    <dependencies>
        
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.version}</artifactId>
            <version>${spark.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-simple</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_${scala.version}</artifactId>
            <version>${spark.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-simple</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>${maven-compiler-plugin.version}</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.codehaus.mojo</groupId>
                <artifactId>exec-maven-plugin</artifactId>
                <version>1.6.0</version>
                <configuration>
                    <executable>java</executable>
                    <arguments>
                        <argument>-classpath</argument>
                        <classpath/>
                        <argument>com.jmb.BasicTransformationsPartTwo</argument>
                    </arguments>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.4.1</version>
                <configuration>
                    
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    
                    <archive>
                        <manifest>
                            <mainClass>com.jmb.BasicTransformationsPartTwo</mainClass>
                        </manifest>
                    </archive>

                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>


</project>

React

Dart

GoJS React

Typescript React

Vue.js


## The scenario

A common scenario while working in batch applications might involve dealing with datasets of information with an odd format, such as the one loaded in the code project initially, We’ll display such datasets with the `show(...)` method call on the DataFrame, which produces the following output:

```
+-------------+-------------+
| Student_Name|       Grades|
+-------------+-------------+
|   John Smith| 7 8 5 8 10 6|
|Patrick Stark|  2 4 2 1 5 6|
| Mary Collins|  7 8 7 7 4 3|
|     Eva Kirk|10 8 9 10 7 6|
|Neil Richards| 6 6 6 8 10 6|
+-------------+-------------+
only showing top 5 rows
```

The file `students_subjects.csv` printed above shows records for students and their grades, but this file's format packs all the grades into one column.



___
Business requirements state that we should process all the data and calculate the subjects to be retaken by replacing the grade with R, for each grade and student, and return this in a new Dataset. If a grade falls below six, then it should be retaken.
___



We can overcome this initial format limitation on the grades column by splitting it into separate elements. These elements come originally separated by a space, so in the code we’ll write:

```
Dataset<Row> studentsDf = df.select(df.col("Student_Name"), split(df.col("Grades"), " "));
```

> **Note:** The method `select(columns...)` is a schema manipulation operation, just as the ones we learned before for adding or dropping columns. In this case, the method creates a new DataFrame with only the selected columns. We can look at this as a method for filtering or keeping/discarding certain columns.

Let's print the output of the `studentsDf`,  resulting from the select operation:

```
+-------------+--------------------+
| Student_Name|split(Grades,  , -1)|
+-------------+--------------------+
|   John Smith| [7, 8, 5, 8, 10, 6]|
|Patrick Stark|  [2, 4, 2, 1, 5, 6]|
| Mary Collins|  [7, 8, 7, 7, 4, 3]|
|     Eva Kirk|[10, 8, 9, 10, 7, 6]|
|Neil Richards| [6, 6, 6, 8, 10, 6]|
| Mary Collins|  [7, 8, 7, 7, 4, 3]|
|Patrick Stark|  [2, 4, 2, 1, 5, 6]|
+-------------+--------------------+
```

We’ve successfully split the Grades column into an array of grades by using the method `split(...)`, which in Spark acts as Spark’s version of the Java split method (split based on a token), but at the column level.

The name of the new column now is not very meaningful, so let’s change it by invoking another schema manipulation function, `withColumnRenamed`.

```
Dataset<Row> renamedStudentsDf = studentsDf.withColumnRenamed("split(Grades,  , -1)", "Grades");
```

It also helps to print the returned DataFrame to check that the transformation was applied correctly:

```
+-------------+--------------------+
| Student_Name|              Grades|
+-------------+--------------------+
|   John Smith| [7, 8, 5, 8, 10, 6]|
|Patrick Stark|  [2, 4, 2, 1, 5, 6]|
| Mary Collins|  [7, 8, 7, 7, 4, 3]|
|     Eva Kirk|[10, 8, 9, 10, 7, 6]|
|Neil Richards| [6, 6, 6, 8, 10, 6]|
| Mary Collins|  [7, 8, 7, 7, 4, 3]|
|Patrick Stark|  [2, 4, 2, 1, 5, 6]|
+-------------+--------------------+
```

Everything seems to look better but our work would be much simpler if we could separate those grades into our own columns and apply the calculation we require.

Luckily for us, the `flatMap` operation exists precisely to solve this problem, so let’s go through the code where we use it.

First, just as we did in the Map and Filter [lesson](https://www.educative.io/courses/mastering-big-data-apache-spark-java-api/transformations-i-map-and-filter) before, we’ll create a separate FlatMapper class and include the processing logic in the `call(...)` method:

```
public class StudentGradesFlatMapper implements FlatMapFunction<Row, Row> {

    @Override
    public Iterator<Row> call(Row row) throws Exception {
        Row newRow = new GenericRowWithSchema(mapAndFlattenArrayValues(row), 
                getStudentsFlattenedSchema());
        return Arrays.asList(newRow).iterator();
    }
```  

The code snippet includes the functionality to:

- Flatten the column that contains an array, and apply the mapping logic to each of its elements. This translates to detecting and flagging which subjects have to be retaken. Because we’re creating a `GenericRowWithSchema` object, we can provide a schema that represents the returned row’s structure in our main class static method. We don’t need to pay too much attention to this for now.


- Return the processed row inside an Iterator of rows.
 

This last part is a bit different than the transformations before, but prior to analyzing why we have to return this object, let’s peek quickly at the method that flattens and flags the grades and highlight some interesting parts:

```
private Object[] mapAndFlattenArrayValues(Row row) {
        WrappedArray arrayColumn = row.getAs("Grades");

        int finalRowSize = arrayColumn.length() + 1;
        Object[] finalRow = new Object[finalRowSize];
        int rowColsIndex = 0;

        //Add first value of row corresponding to name
        finalRow[rowColsIndex++] = row.getAs("Student_Name");

        //Loop through the column 2 values within the array and add them as separate columns
        String[] arrayColumnValues = (String[]) arrayColumn.array();
        for (int i = 0; i < arrayColumnValues.length; i++) {
            //check if grades has to be retaken
            String currentGrade = arrayColumnValues[i];
            if (Integer.valueOf(arrayColumnValues[i]) < 6) {
                currentGrade = "R";
            }

            finalRow[rowColsIndex++] = currentGrade;
        }
        return finalRow;
    }
```

The main logic is easy enough to understand. The interesting parts are:

- Retrieving the column with the grade's array, by invoking `WrappedArray arrayColumn = row.getAs("Grades");`.

- Creating an array of objects to represent the returned row columns, which includes the Student_Name Column plus all the columns present in the array of grades.

```
int finalRowSize = arrayColumn.length() + 1;
        Object[] finalRow = new Object[finalRowSize];
```

- Processing each grade and including it in the `finalRow` array, within the `for` loop (see above main code snippet).

Back in the main class, this logic is invoked as follows:

```
        Dataset<Row> flattenedDf = renamedStudentsDf
                .flatMap(new StudentGradesFlatMapper(), RowEncoder.apply(getStudentsFlattenedSchema()));
        flattenedDf.show();
```

The last line prints the results, where we can observe that the flattening and mapping took place as desired:

```
+-------------+-------+-------+-------+-------+-------+-------+
|         Name|Grade 1|Grade 2|Grade 3|Grade 4|Grade 5|Grade 6|
+-------------+-------+-------+-------+-------+-------+-------+
|   John Smith|      7|      8|      R|      8|     10|      6|
|Patrick Stark|      R|      R|      R|      R|      R|      6|
| Mary Collins|      7|      8|      7|      7|      R|      R|
|     Eva Kirk|     10|      8|      9|     10|      7|      6|
|Neil Richards|      6|      6|      6|      8|     10|      6|
| Mary Collins|      7|      8|      7|      7|      R|      R|
|Patrick Stark|      R|      R|      R|      R|      R|      6|
+-------------+-------+-------+-------+-------+-------+-------+
```


Why the need to return an Iterator of rows?

It is needed because the interface we implemented in our FlatMapper specifies the following on its contract:

```
@FunctionalInterface
public interface FlatMapFunction<T, R> extends Serializable {
    Iterator<R> call(T var1) throws Exception;
}
```

From a functional point of view, it provides us with a way of working with one or all columns representing collections of elements that we might want to move into separate rows.

As a quick example, imagine that we have a row with a column like this:

`[{POJO1, POJO2, ... POJON}]`

The column is a collection itself, so we might need to expand it to a row, each containing a POJO (and its properties), such as:

`Row 1: POJO1` `(name: aName, age: anAge, etc)`

`Row 2: POJO2` `(name: aName, age: anAge, etc)`

For this to be possible, we would need to return each of these POJOs in a collection, or in an `Iterator<Row>`, which is what Spark requires us to return in the Interface.




# Distinct

The **Distinct** transformation removes duplicated rows from a DataFrame, just as the SQL distinct operation does on rows of a table.


Get introduced to the second set of basic transformations.

Course Introduction

Spark Introduction and Basics

Getting Started with Spark

DataFrame Basic Operations

DataFrame Advanced Operations

Spark SQL and Other Functionalities

Building a Big Data Batch Application

Deployment and Cluster Execution

Monitoring and Performance Fundamentals

Conclusion

Apendix

Transformations (II): FlatMap and Distinct

FlatMap