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

# Union

The `union` transformation allows us to combine two DataFrames, thus producing a new one containing the rows from both.

This operation has the following characteristics:

- The schemas of both DataFrames have to be identical. This doesn’t detour much from the classical SQL UNION operation available in RDBMS.


- Duplicate records are preserved and aggregated to the final results.


We are going to first present a graphical representation of this transformation, which illustrates an interesting property that makes `union` an attractive transformation in specific scenarios.




The `union` transformation merges and piles up one DataFrame after the another. No exchange of information happens between them or, more importantly, their partitions, which are omitted for this reason on the diagram. Also, no sorting is applied before merging the rows.

This fundamentally means that no shuffling occurs in this transformation, and the rows are just collected as they come into a new DataFrame.

> **Note:** Rows are not actually collected as  with the `collect()` action and returned back to the driver program. Instead, the operation to perform a union re-assigns the rows to a new DataFrame. In other words, if a DataFrame is an abstraction of rows in the cluster nodes, the result of this transformation is an abstraction of a DataFrame with more rows than the original two DataFrames.


The business domain for this lesson’s project contains information about sales people working for a fictitious company. This company, in turn, owns a couple of offices: a main branch in the United States and a branch in London, in the United Kingdom.


##############################
## 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 Union and DropDuplicates Transformations Example Project

Project demonstrates usage of the `Distinct` and `Union` transformations within the *Spark* API:

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>UnionAndDistinct</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.UnionAndDistinct</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.UnionAndDistinct</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

> **Note:** Due to this transformation outputting a considerable amount of logs (just like in real-life scenarios), we’ve included the following breakpoint messages that require input to proceed with program execution at different places: “Input any non-blank key and tap Enter…” These help to show the results of each transformation separately and in a more interactive way.



A simple requirement involves listing the totality of employees based on two datasets provided by the respective offices’ HR departments. Because both branches use a unified HR system, the file formats are the same, meaning they include the same amount of fields and types, and so on.

The scenario presents a golden opportunity to use the `union(...)` transformation and merge these datasets of information together.

Two CSV files are read into separate DataFrames and their schemas, or structures, are inspected by calling the `printSchema()` method on both:

```
dfSalesUs.printSchema();
dfSalesUk.printSchema();
```

The output looks like this:

```
root
 |-- employee_name: string (nullable = true)
 |-- department: string (nullable = true)
 |-- salary: double (nullable = true)

root
 |-- employee_name: string (nullable = true)
 |-- department: string (nullable = true)
 |-- salary: double (nullable = true)
```

A DataFrame holds five records for the US:

```
+---------------+--------------+-------+
|  employee_name|    department| salary|
+---------------+--------------+-------+
|     John Smith|         books|10000.0|
|     Juan Perez| digital media| 4000.0|
|Lorraine Brooks|  social media| 5000.0|
|  Richard Massi|    management|20000.0|
| Augustus Blake|  social media| 4500.0|
+---------------+--------------+-------+
```

And the second DataFrame holds five for the UK as well:

```
+--------------+--------------+-------+
| employee_name|    department| salary|
+--------------+--------------+-------+
| Richard Massi|    management|15000.0|
|Augustus Blake|  social media| 4020.0|
|   Brett Hicks|         books|11000.0|
| Emanuel Gomez| digital media| 4100.0|
| Robert Doyle |  social media| 3900.0|
+--------------+--------------+-------+
```

The signature for `union` is explicit and easy to understand:

```
public Dataset<T> union(final Dataset<T> other)
```

The method is called on a DataFrame while passing a second DataFrame on which to perform the `union` operation.:

```
Dataset<Row> allEmployees = dfSalesUs.union(dfSalesUk);
``` 

> **Note:** The order of the DataFrames involved in the union transformation produces a different order of rows.


We can display all the rows of the newly produced `allEmployees` DataFrame by invoking the `show()` action:

```
+---------------+--------------+-------+
|  employee_name|    department| salary|
+---------------+--------------+-------+
|     John Smith|         books|10000.0|
|     Juan Perez| digital media| 4000.0|
|Lorraine Brooks|  social media| 5000.0|
|  Richard Massi|    management|20000.0|
| Augustus Blake|  social media| 4500.0|
|  Richard Massi|    management|15000.0|
| Augustus Blake|  social media| 4020.0|
|    Brett Hicks|         books|11000.0|
|  Emanuel Gomez| digital media| 4100.0|
|   Robert Doyle|  social media| 3900.0|
+---------------+--------------+-------+
```



# UnionByName

If we want to merge information from DataFrames with the same columns but in different positions (schema wise), we can use the `unionByName` transformation.

However, all the constraints mentioned before for `Union` apply to this transformation as well. This includes constraints such as having the same number of columns, the same column types, and so on.


To test the transformation, we drop the “department” column on the UK DataFrame and create a new one by adding it back at the end (default behavior when adding a column) using the `withColumn(...)` transformation.

We also fill the new column values with a literal string, `“TBD”`:

```
Dataset<Row> dfUkSalesReOrg = dfSalesUk.drop("department").withColumn("department", lit("TBD"));
```

If we look at the first two entries, we can see the changes reflected on the new DataFrame, that is, the "Department" column for the UK DataFrame is now the last column:

```
+--------------+-------+----------+
| employee_name| salary|department|
+--------------+-------+----------+
| Richard Massi|15000.0|       TBD|
|Augustus Blake| 4020.0|       TBD|
+--------------+-------+----------+
only showing top 2 rows
```

The `unionByName` function is used in the following manner:

```
//Now attempt a unionByName - union based on matching columns' Names, regardless of their schemas
        Dataset<Row> allEmployeesUnionByName = dfSalesUs.unionByName(dfUkSalesReOrg);
```

By using this transformation, we see the same results being printed:

```
+---------------+--------------+-------+
|  employee_name|    department| salary|
+---------------+--------------+-------+
|     John Smith|         books|10000.0|
|     Juan Perez| digital media| 4000.0|
|Lorraine Brooks|  social media| 5000.0|
|  Richard Massi|    management|20000.0|
| Augustus Blake|  social media| 4500.0|
|  Richard Massi|           TBD|15000.0|
| Augustus Blake|           TBD| 4020.0|
|    Brett Hicks|           TBD|11000.0|
|  Emanuel Gomez|           TBD| 4100.0|
|   Robert Doyle|           TBD| 3900.0|
+---------------+--------------+-------+
```



# DropDuplicates

If memory serves us well, when we learned the `distinct()` transformation in past lessons, we learned it mandates that two or more rows should have exactly the same values for every column, to work properly, much like the traditional `SQL DISTINCT` clause in RDBMS systems. This identity condition is not enforced on the dropDuplicates(...) transformation.


Instead, the identity condition to detect duplicates is specified by passing into the method columns’ names.

Let’s start by visualizing this concept before inspecting the code example.




Get introduced to Union, UnionByName, and DropDuplicates transformations in this lesson.

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

Union, UnionByName, and DropDuplicates

Union