Spark SQL

This module works well with structured data, meaning you can read data stored in an RDBMS table or from files with structured data in CSV, Text, JSON, Avro, ORC, or Parquet format and then construct permanent or temporary tables in Spark. Also, when using Spark’s Structured APIs in Java, Python, Scala or R, you can combine SQL-like queries to query the data just read into a Spark DataFrame.¹⁸ To date, Spark SQL is ANSI SQL:2003 compliant.¹⁹

For example, in this Scala code snippet, you can read from a JSON file stored on Amazon S3, create a temporary table, and issue a SQL-like query on the results read into memory as a Spark DataFrame.

Note: You can issue the same code snippet in Python, R or Java, and the generated bytecode will be identical, resulting in the same performance.

// read data off Amazon S3 bucket into a Spark DataFrame
            spark.read.json("s3://apache_spark/data/committers.json").createOrReplaceTempView("committers")
            // issue an SQL query and return the result as a Spark DataFrame
            val results = spark.sql("""SELECT name, org, module, release, num_commits
            FROM committers WHERE module = ‘mllib’ AND num_commits > 10 
            ORDER BY num_commits DESC""")

Spark MLlib

Spark comes with a library containing common classical machine learning (ML) algorithms called MLlib. Since Spark’s first release, the performance of this library component has improved because of Spark 2.x’s underlying engine enhancements. Today, MLlib provides data scientists with a popular list of machine learning algorithms, including classification, regression, clustering, collaborative filtering, build atop high-level DataFrame-based APIs to build predictive models.

These APIs allow you to do featurization, to extract or transform features, to build pipelines (for training and evaluating), and to persist models (for saving and reloading them) during deployment. Additional utilities include the use of common linear algebra operations and statistics. MLlib includes other low-level ML primitives, including a generic gradient descent optimization. Using just the high-level DataFame-based APIs, this Python code snippet encapsulates the basic operations a typical data scientist may do when building a model.

Note: Starting Apache Spark 1.6, the MLlib project split between two packages: spark.mllib and spark.ml. The latter is DataFrame-based APIs while the former is RDD-based APIs, which is now in maintenance mode. All new features and development by the community goes into spark.ml²⁰

This code snippet is just an illustration of the ease with which you can build your models—it feels Pythonic. More extensive examples will be discussed in chapters 11 and 12.

from pyspark.ml.classification import LogisticRegression
            ...
            training = spark.read.csv("s3://...")
            test = spark.read.csv("s3://...")
            
# Load training data
lr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)
            
# Fit the model
lrModel = lr.fit(training)
            
# predict
            lrModel.transform(test)
...

All these new enhancements in spark.ml are designed for developer’s ease of use and for scaling training across the Spark cluster.

Spark Structured Streaming

Apache Spark 2.0 introduced an experimental Continuous Streaming model²¹ and Structured Streaming APIs²², built atop the Spark SQL engine and DataFrame-based APIs. By Spark 2.2, Structured Streaming was generally available, meaning that developers could use it in their production environments.

Necessary for big data developers to combine and react in real-time to both static data and streaming data from engines like Apache Kafka and other streaming sources, the new model viewed a stream as a continually growing table, with new rows of data appended at the end. And developers merely treated this as a structured table and issued queries against it as they would a static table.

This new model obviated the old DStreams model in Spark’s 1.x series, which we will discuss in more detail in chapter 9. Underneath the structured streaming model, the Spark SQL core engine handles all aspects of fault-tolerance and late-data semantics, allowing developers to focus on writing streaming applications with relative ease.

Furthermore, Spark 2.x extends its streaming data sources to include Apache Kafka, Kinesis, and HDFS-based or cloud storage.

For example, this code snippet shows a typical anatomy of a Structured Streaming application doing the Hello World “word count” of a streaming application.

Note: Take note of the composability of the high-level domain specific language (DSL) syntax of the DataFrame-based APIs transformation and computation. It has the look-and-feel of SQL-like aggregation but done in Python DataFrame API. We will expand on this in the later chapters.

# read from Kafka stream
            lines = (spark.readStream 
             .format("kafka") 
             .option("subscribe", "input") 
             .load())
            # perform transformation
            wordCounts = (lines.groupBy("value.cast(‘string’) as key") 
             .agg(count("*") as "value") )
            # write back out to the stream
            query = (wordCounts.writeStream() 
             .format("kafka") 
             .option("topic", "output")) 

GraphX

As the name suggests, GraphX is a library for manipulating graphs (e.g., social networks graphs, routes and connections points, network topology) and performing graph-parallel computations. Contributed by many users in the community, it offers the standard graph algorithms for analysis, connections, and traversals. Algorithms include PageRank, Connected Components, and Triangle Counting. This code snippet shows a simple example of how to join two graphs using the GraphX APIs.

val graph = Graph(vertices, edges)
            messages = spark.textFile("hdfs://...")
            graph2 = graph.joinVertices(messages) {
             (id, vertex, msg) => ...
            }

Spark Session

In Spark 2.0, the SparkSession became a unified conduit to all Spark operations and data. Not only did it subsumes previous entry points to Spark like SparkContext, SQLContext, HiveContext, SparkConf and StreamingContext, but it also made working with Spark simpler and easier. ²³

Because of this unified conduit, you can create JVM runtime parameters, create DataFrames and Datasets, read from data sources, access Catalog Metadata, and issue Spark SQL queries. This is the essence of simplicity in Spark 2.x: a single unified entry point to all of Spark’s functionality!

In a standalone Spark application, you can create a SparkSession using one of the high-level APIs in the programming language of your choice, whereas in Spark shells (more on this later in the next chapter), it’s created for you, and you can access it via a global variable called spark or sc.

For example, in a Spark application, you can create a SparkSession per JVM and use it to perform a number of Spark operations, whereas in Spark 1.x, you would have had to create individual contexts, introducing extra boilerplate code.

Note: Although in Spark 2.x, the SparkSession subsumes all other contexts, you can still access the individual contexts and its respective methods. This way, the community maintained backward compatibility. That is, your old 1.x code with SparkContext or SQLContext will still work.

import org.apache.spark.sql.SparkSession
            // build SparkSession
            val spark = SparkSession().builder()
            .appName("LearnSpark")
            .config("spark.sql.shuffle.partitions", 6)
            .getOrCreate()
            ...
            val people = spark.read.json("...")
            ...
            val resultsDF = spark.sql("SELECT city, pop, state, zip FROM zips_table")

Spark Driver

As part of the Spark application responsible for instantiating a SparkSession, the Spark Driver has multiple responsibilities: it communicates with the Cluster Manager (explained below); it requests resources from the Cluster Manager for Spark’s Executors JVMs; and it transforms all the Spark operations into DAG computations, schedules them, and distributes their execution as tasks across all Spark Executors.

Its interaction with the Cluster Manager is merely to get Spark Executor resources: JVMs, CPU, memory, disk etc. Once allocated, it communicates directly with Spark Executors.

Cluster Manager

The Cluster Manager is responsible for managing and allocating resources for a cluster of nodes on which your Spark application runs, regardless of the deployment modes (see the Table 1-1). Sometimes called Spark or Cluster “workers,” these managed cluster nodes launch individual Spark Executors. Currently, Spark supports four cluster managers: a built-in standalone cluster manager, Apache Hadoop YARN, Apache Mesos, and Kubernetes.

Spark Executor

The Spark Executor runs on each “worker” node in the cluster. Often called a Spark “worker,” it is responsible to launch Executors on which Spark’s tasks run. Usually, only a single Executor runs per node in various deployment modes.

Deployment Modes

An attractive feature of Spark is its support for myriad deployment modes and in various popular environments. Because the cluster manager is agnostic to where they run (as long as they can manage Spark workers’ Executors and fulfill resource requests), Spark can be deployed in some of the most popular environments -- such as Apache Mesos, Apache Hadoop YARN, and Kubernetes -- and can operate in different modes.

Mode	Spark Driver	Spark Executor	Cluster Manager
Local	Runs on a single JVM, like a laptop and a single node.	Runs on the same JVM as the driver	Runs on the same host.
Standalone	Can run on any node in the cluster	Each node in the cluster will launch its own Executor JVM	Can be allocated arbitrarily to any host in the cluster
YARN (client)	On a client, not part of the cluster	YARN’s NodeManager’s Container	YARN’s Resource Manager works with YARN’s Application Master to allocate the containers on NodeManagers for Executors
YARN (cluster)	Runs with the YARN’s Application Master	Same as YARN client mode	Same as the YARN client mode
Mesos (client)	Runs on a client, not part of Mesos cluster	Container within Mesos Agent	Mesos’ Master
Mesos (cluster)	Runs within one of Mesos’ Master	Same as client mode	Mesos’ Master
Kubernetes	Kubernetes pod	Each worker runs within its own pod	Kubernetes Master

Table 1-1 Cheat Sheet for Spark Deployment modes ²⁴

Distributed Data and Partitions

Actual physical data is distributed across an entire Spark cluster as partitions residing in either HDFS or cloud storage. Often more than one partition may reside on a node in a cluster.

While the data is distributed as partitions across the physical cluster, Spark treats them as a high-level logical data abstraction -- as an RDD or DataFrame in memory -- where each Spark worker’s Executor preferably (though not always possible) is allocated a task that can read the partition closest to it in the network (observing data locality). See Figure. 1-5.

Figure 1-5. Data distributed across physical machines

Partitioning allows for efficient parallelism. A distributed scheme of breaking up data into chunks of partitions allows Spark Executors to process only data that is close to them, minimizing network bandwidth. That is, each Executor’s core or slot is assigned its own data partition to work on (Fig. 1-6).

For example, this code snippet will break up the physical data stored across clusters into 8 partitions, and each Executor will get some partitions to read into its memory.

log_df = spark.read.text(“path_to_large_text_file”).repartition(8)
            println(log_df.rdd.getNumPartitions())>
            Or this code will create a DataFrame of 10,000 integers distributed over 8 partitions in memory.
            df = spark.range(0, 10000, 1, 8)
            println(df.rdd.getNumPartitions())

Both code snippets will print out 8.

Figure 1-6. Each Executor’s core gets a partition of data to work on

In chapters 3 and chapter 8, we will discuss how to tune and change partitioning configuration for maximum parallelism based on how many cores (or slots) you have on your Executors.