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

# Sharing data in a cluster

Sharing data in a distributed environment, regardless of the use case, can be confusing.

Understanding the scope (where the variables “live”) and lifecycle (how the values change) of shared variables while executing code in a cluster presents itself as a challenging task.


Within the Spark ecosystem, variables can be passed down to objects that operate in a distributed fashion. Still, these are copies with a different state each while execution takes place.

Furthermore, this is a one-way type of communication, meaning that those variables are not sent back to the driver program with whatever updated values they might have.

To address these inconveniences, the Spark API provides both **accumulators** and **broadcast variables**.

## Accumulators

Accumulators are variables that expose only an addition operation. This means we can only add values but not delete, or modify existing values. In other words, accumulators provide a simple way of aggregating values from the worker nodes back to the driver program.

Let’s start with a code example to illustrate. Let’s work on a brief project to demonstrate the usage of accumulators and broadcast variables.



##############################
## 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 Accumulators and Broadcast Variables Example Project

Project demonstrates usage of the `Accumulators` construct and `Broadcast Variables` 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>Accumulators-broadcasting</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.AccumulatorsBroadcasting</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.AccumulatorsBroadcasting</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 business domain of this application deals with network information, specifically network’s transit data concerning peer-to-peer communication from low-level devices, such as a series of transmitters in a hardware device.

The data is produced frequently as a CSV file and analyzed to isolate and potentially replace slow network paths caused by a faulty transmitter (by inspecting the “ping” of each communication). The information is loaded in a DataFrame with a schema as follows:

```
root
 |-- source: string (nullable = true)
 |-- destination: string (nullable = true)
 |-- ping: double (nullable = true)
 |-- errors: integer (nullable = true)
```

We can imagine a vast amount of rows received per file because communications within these devices run in the thousands per second. For our demonstration purposes, we can just work with a tiny dataset of only six rows.

So, we decide to implement the `MapFunction<>` interface from Spark to work on each row and produce a new DataFrame containing transmitters’ IDs, which are the first two letters of a source or destination.

Then, we tag them according to a category based on slowness tolerance such as:

- Ping <= 50: ACPT (Acceptable)
- Ping 50 < ping <= 100: CGTD (Congested)
- Ping > 100: OVLD (Overloaded)


Fundamentally, we want to produce rows with the following schema:

```
+--------------+--------+
|Transmitter_id|Category|
+--------------+--------+
```

Our `Mapper` class does the heavy lifting, and the `call(...)` method that inspects a row and produces a new, categorized row looks like this:

```
public class TransmitterSpeedMapper implements MapFunction<Row, Row> { 

 //Other Code...

    @Override
    public Row call(Row row) throws Exception {
        //Get ping, analyze and produce new row
        int rowPing = row.getInt(pingColId);
        String transId = row.getString(sourceColId);
        String category = getCategory(rowPing);
        return produceRow(transId, category);
    }
```

The logic is quite simple and contains the following sequential steps:

1. Get the needed information from the row received (the ping and transmitter ID) and classify it according to the categories defined before.


2. Produce the expected row with the targeted schema.


> **Note:** Most of the “collaborator” methods of this class (private methods) are omitted, but is highly advisable to read them and try to understand what functionality they provide.


We put our code to the test by invoking the `Mapper` class on the original DataFrame:

```
//Analyze the information with the Mapper and produce the new DataFrame
Dataset<Row> results = df.map(new TransmitterSpeedMapper(),
                RowEncoder.apply(TransmitterSpeedMapper.structType()));
```

We trigger then the transformation with the `show()` action on the `results` DataFrame, and an exception is thrown:

```
21/03/06 18:36:40 ERROR Executor: Exception in task 0.0 in stage 2.0 (TID 2)
java.lang.NullPointerException: Value at index 2 is null
	at org.apache.spark.sql.Row.getAnyValAs(Row.scala:514)
	at org.apache.spark.sql.Row.getInt(Row.scala:236)
	at org.apache.spark.sql.Row.getInt$(Row.scala:236)
	at org.apache.spark.sql.catalyst.expressions.GenericRow.getInt(rows.scala:166)
	at com.jmb.mapper.TransmitterSpeedMapper.call(TransmitterSpeedMapper.java:31)
```

It seems there are empty rows or rows with `null` values in the file received, which is valuable information around potential problems with a device.

We can count the number of rows with erroneous or blank information, by dealing with the `null` values present in the DataFrame per row.

If the counter of rows with problems is too high, we can always issue a device replacement order.

We can first try passing a `counter` variable to the class constructor in the `Mapper`:

```
Integer errorsCounter = 0;
Dataset<Row> results = df.map(new TransmitterSpeedMapper(errorsCounter),
                RowEncoder.apply(TransmitterSpeedMapper.structType()));
```

To try these changes, we include extra logic that detects if there are any `null` values per row, and if there are, it produces an error row. It also counts the occurrences, so then we don’t have to count these erroneous rows by firing another transformation. This is no trivial detail—the number of rows is huge, and this saves precious time and cost.

The code additions are written as:

```
//Constructor receives counter
public TransmitterSpeedMapper(int errCounter) {
     this.errCounter = errCounter;
}

//Code to deal with erroneous rows
if(row.anyNull()) {
     errCounter++;
     return produceRow(ERR_VAL, ERR_VAL);
}
```

Finally, back in the driver program code, we can evaluate the number of erroneous rows processed:

```
//Print errorsCounter variable
LOGGER.info("Amount of rows with errors: " + errorsCounter);
```

This displays the following in turn:

```
INFO AccumulatorsBroadcasting: Amount of rows with errors: 0
```

This didn’t work as expected. Why? Recall the following: Spark works distributedly, in different nodes that run different processes and within different JVMs.

So, not only does each node receive a different counter, but it also can’t send it back as the same variable. Consequently, it doesn’t accumulate all the nodes’ counters.

The good news is that Spark accumulators are just the tool we can use to solve this problem, so let’s see this in action.

First, we retrieve an `Accumulator` object from the `SparkContext`, which handles the synchronization of these distributed objects for us. Then, we use it in conjunction with one of the Spark API Functions (`ForEachFunction`) as shown in the below code snippet:

```
//Register and use and accumulator
LongAccumulator longAccumulator = session.sparkContext().longAccumulator("ErrorsCounter");
        df.foreach((ForeachFunction<Row>) row -> {
            if(row.anyNull()){
                longAccumulator.add(1L);
            }
});
```

Right after, we print the results and can see them accumulated:

```
//Print the errors accumulator
LOGGER.info("Amount of rows with errors: " +  longAccumulator.value());
```

Here’s the output:

```
INFO AccumulatorsBroadcasting: Amount of rows with errors: 1
```

That now seems to be working as expected. On top of that, if we print the processed rows with a `results.show()`, we can observe the erroneous rows are present as well:

Here’s the output:

```
+--------------+--------+
|Transmitter_id|Category|
+--------------+--------+
|          ans1|    OVLD|
|          ans1|    ACPT|
|           ERR|     ERR|
|          xty3|    ACPT|
|           aaa|    CGTD|
|              |    CGTD|
+--------------+--------+
```



### Collections’ accumulators

Accumulators backed by collections are another flavor provided by the Spark API.

A potential use case for them, one that pertains to our example’s business domain, is to add all the transmitters’ IDs of rows with a ping value of zero, which also represents an anomaly during the transmissions.


We can implement it with the following code, which makes use of the `CollectionAccumulator` Spark class:

```
CollectionAccumulator<PingAnomaly> pingAccum = session.sparkContext().collectionAccumulator();

int sourceColId = 0;
int destColId = 1;
int pingColId = 2;

df.foreach((ForeachFunction<Row>) row -> {
        if(!row.anyNull() && row.getInt(pingColId) == 0){
         pingAccum.add(new PingAnomaly(row.getString(sourceColId), row.getString(destColId)));
        }
});
```

The following remarks about the code are important:

 - A new DTO class, `PingAnomaly` is created, which acts as a “placeholder class” that the `CollectionAccumulator` can work with. It targets the properties we are interested in retrieving and pushing to the `Accumulator`. This is why we type the `Accumulator` to it.


- The `Accumulator` is used in the same way as before with the `LongAccumulator`. The only difference is in the value that we add to it. This allows for a bit of a somewhat more refined logic while adding accumulated values.


The output when we run the program shows the following content:

```
21/03/08 16:29:07 INFO AccumulatorsBroadcasting: Amount of rows with anomalies: 
21/03/08 16:29:07 INFO AccumulatorsBroadcasting: Source - dest xty3-xtq1
```

### Accumulator observations

There is an interesting discussion to be had around the need to use of accumulators and how reliable they are. Consider the below points:

- Spark transformations, such as aggregated functions or `Reduce`, can provide an accumulated value with the benefit that they are more explicit and easy to understand code-wise, and allows us to produce complex types as the output.
- In a distributed environment, because it is not transparent to the developer how Spark manages the processes, the order of the data accumulated cannot be predicted.
- If there are any failures, the accumulator values might not be accurate. Even if the flow of the program resumes and tasks are re-processed, duplicated values can be accumulated.

Ultimately the business cases of our application might or might not provide the need to use accumulators. The good news is that most, if not all, things that can be achieved with an accumulator can also be done with other transformations.

## Broadcast Variables

Broadcast variables share the notion of exposing a variable from the driver program to the nodes where the executors are running the transformations, but differ from accumulators in that they are read-only.

The main benefit of using these types of variables is that they can help reduce the amount of static or immutable data that executors might need, thus minimizing network overhead and memory footprint. Instead of providing a copy of data to each process running in parallel, a broadcast variable can be shared.

Let’s dive into an example.

Imagine we have a data structure representing a lookup table for transmitter manufacturers, initially retrieved from a database. Instead of sending a copy to each process, we can just declare it as a `Broadcast` variable and re-use it within our `Mapper` class.


We can then filter the transmitters with a lot of latency (high ping) and attach a column with the manufacturers’ information to fulfill future replacements.

The code to achieve this looks like this:

```
//Filter transmitters with high ping as depicted by the category OVLD
Dataset<Row> slowTransmitters = results.filter((FilterFunction<Row>) row -> row.getAs("Category").equals("OVLD")
);
```

The code snippet above filters the transmitter with the sought-after category for the requirement at hand.

A lookup table is defined, based on a Java `Map` object, and registered as a `Broadcast` variable, which is achieved through the `SparkSession` and `SparkContext`:

```
//Declare the lookup table as Broadcast Variable
Map<String, String> lookupTable = new HashMap<>();
lookupTable.put("ans","American Networking");
lookupTable.put("xty","Expert Transmitters");
lookupTable.put("xfh","XFH");

Broadcast<Map<String, String>> bcVar = session.sparkContext().broadcast(lookupTable,  scala.reflect.ClassTag$.MODULE$.apply(Map.class));

```

All that remains is to trigger the `Mapper` class that uses the `Broadcast` variable:

```
//Apply a map transformation using the broadcast variable
TransmitterCompanyMapper companyMapper = new TransmitterCompanyMapper(bcVar);
Dataset<Row> manufacturers = slowTransmitters.map(companyMapper, RowEncoder.apply(companyMapper.structType()));

```

It’s important to highlight how the variable value is retrieved on the `Mapper class` before it can be used:

```
@Override
public Row call(Row row) {
//other code...
lookupTable.getValue().get(transId);
//code continues...
```

The final DataFrame output looks how we want it to:

```
+--------------+-------------------+
|Transmitter_id|       Manufacturer|
+--------------+-------------------+
|           ans|American Networking|
+--------------+-------------------+
```




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

Accumulators and Broadcast Variables

Sharing data in a cluster