Notes.txt

GENERAL INTRODUCTION INTO SPARK-

Apache Spark is a unified computing engine and a set of libraries for parallel data processing on
computer clusters. Spark is the most actively developed open source engine for this task, making
it a standard tool for any developer or data scientist interested in big data. Spark supports 
multiple widely used programming languages (Python, Java, Scala, and R), includes libraries for
diverse tasks ranging from SQL to streaming and machine learning, and runs anywhere from a laptop
to a cluster of thousands of servers. This makes it an easy system to start with and scale-up 
big data processing or incredibly large scale.

Spark is designed to support a wide range of data analytics tasks, ranging from simple data 
loading and SQL queries to machine learning and streaming computation, over the same computing 
engine and with a consistent set of APIs. 


PROPERTIES-
1. Spark’s unified nature makes these tasks both easier and more efficient to write. First, Spark
provides consistent, composable APIs that you can use to build an application out of smaller
pieces or out of existing libraries. It also makes it easy for you to write your own analytics
libraries on top. Spark’s APIs are also designed to enable high performance by optimizing across
the different libraries and functions composed together in a user program. 

For example, if you load data using a SQL query and+
then evaluate a machine learning model over it using Spark’s ML library, the engine can
combine these steps into one scan over the data. The combination of general APIs and high
performance execution, no matter how you combine them, makes Spark a powerful platform
for interactive and production applications.

2. Spark handles loading data from storage systems and
performing computation on it, not permanent storage as the end itself. You can use Spark
with a wide variety of persistent storage systems, including cloud storage systems such as
Azure Storage and Amazon S3, distributed file systems such as Apache Hadoop, key-value
stores such as Apache Cassandra, and message buses such as Apache Kafka. However, Spark
neither stores data long term itself, nor favors one over another. The key motivation here is
that most data already resides in a mix of storage systems. Data is expensive to move so
Spark focuses on performing computations over the data, no matter where it resides. In userfacing 
APIs, Spark works hard to make these storage systems look largely similar so that
applications do not need to worry about where their data is.

3. Spark’s final component is its libraries, which build on its design as a unified engine to
provide a unified API for common data analysis tasks. Spark supports both standard libraries
that ship with the engine as well as a wide array of external libraries published as third-party
packages by the open source communities. Today, Spark’s standard libraries are actually the
bulk of the open source project: the Spark core engine itself has changed little since it was
first released, but the libraries have grown to provide more and more types of functionality.
Spark includes libraries for SQL and structured data (Spark SQL), machine learning (MLlib),
stream processing (Spark Streaming and the newer Structured Streaming), and graph
analytics (GraphX). Beyond these libraries, there are hundreds of open source external
libraries ranging from connectors for various storage systems to machine learning algorithms.


SPARK ARCHITECTURE

Single machines do not have enough power and resources to perform
computations on huge amounts of information (or the user probably does not have the time to
wait for the computation to finish). A cluster (or group of computers), pools the resources of
many machines together, giving us the ability to use all the cumulative resources as if they were
a single computer. Now, a group of machines alone is not powerful, you need a framework to
coordinate work across them. Spark does just that, managing and coordinating the execution of
tasks on data across a cluster of computers.


Spark Applications consist of 
(i): a driver process.
(ii): a set of executor processes.

The driver process runs your main() function, sits on a node in the cluster, and is responsible 
for three things: maintaining information about the Spark Application; responding to a user’s program or input;
and analyzing, distributing, and scheduling work across the executors.  It maintains all
relevant information during the lifetime of the application.

The executors are responsible for actually carrying out the work that the driver assigns them.
This means that each executor is responsible for only two things: executing code assigned to it
by the driver, and reporting the state of the computation on that executor back to the driver node.

In shorts, Spark employs a cluster manager that keeps track of the resources available. The driver process 
is responsible for executing the driver program’s commands across the executors to complete a given task. 


STARTING SPARK
When you start Spark in this interactive mode, you implicitly create a SparkSession that manages
the Spark Application. When you start it through a standalone application, you must create the
SparkSession object yourself in your application code. The SparkSession instance is the way Spark executes
user-defined manipulations across the cluster. There is a one-to-one correspondence between a
SparkSession and a Spark Application. 
  

In Python, the variable is available as "spark" when you start the console.

>>>> spark

Let’s now perform the simple task of creating a range of numbers. This range of numbers is just
like a named column in a spreadsheet:

>>>> myRange = spark.range(1000).toDF("number")

Here above, we created a DataFrame with one column containing 1,000
rows with values from 0 to 999. This range of numbers represents a distributed collection. When
run on a cluster, each part of this range of numbers exists on a different executor. This is a Spark
DataFrame.


DATAFRAMES-
A DataFrame is the most common Structured API and simply represents a table of data with
rows and columns. A spreadsheet sits on one computer in one specific
location, whereas a Spark DataFrame can span thousands of computers.

Partitions-
To allow every executor to perform work in parallel, Spark breaks up the data into chunks called
partitions. A partition is a collection of rows that sit on one physical machine in your cluster. A
DataFrame’s partitions represent how the data is physically distributed across the cluster of
machines during execution. If you have one partition, Spark will have a parallelism of only one,
even if you have thousands of executors. If you have many partitions but only one executor,
Spark will still have a parallelism of only one because there is only one computation resource.
With DataFrames you do not (for the most part) manipulate
partitions manually or individually. You simply specify high-level transformations of data in the
physical partitions, and Spark determines how this work will actually execute on the cluster.


Transformations-
In Spark, the core data structures are immutable, meaning they cannot be changed after they’re
created. To “change” a DataFrame, you need to instruct Spark how you would like to
modify it to do what you want. These instructions are called transformations.

# in Python
>>>> divisBy2 = myRange.where("number % 2 = 0")

Notice that these return no output. This is because we specified only an abstract transformation,
and Spark will not act on transformations until we call an action.

Transformations are the core of how you express your business logic using Spark. There are two
types of transformations: those that specify narrow dependencies, and those that specify wide
dependencies.

Transformations consisting of narrow dependencies (we’ll call them narrow transformations) are
those for which each input partition will contribute to only one output partition. In the preceding
code snippet, the where statement specifies a narrow dependency, where only one partition
contributes to at most one output partition.

A wide dependency (or wide transformation) style transformation will have input partitions
contributing to many output partitions. This is also referred as 'Shuffle'.


____________________________
EXTRAS

Transformation:
Transformations is a kind of process that will transform your RDD data from one form to another 
in Spark. and when you apply this operation on an RDD, you will get a new RDD with transformed data (RDDs in Spark are immutable).

Transformations are the core of how you will be expressing your business logic using DAG in Spark and 
intermediate operation conversion from one form to another RDD.
Two types of transformations in SPARK:
1.Wide Transformations
2.Narrow Transformations

Let’s see one by one.


Narrow Transformations:
These types of transformations convert each input partition to only one output partition. 
When each partition at the parent RDD is used by at most one partition of the child RDD or when each partition 
from child produced or dependent on single parent RDD.

This kind of transformation is basically fast.
Does not require any data shuffling over the cluster network or no data movement.
Operation of map()and filter() belongs to this transformations.

Wide Transformations:
This type of transformation will have input partitions contributing to many output partitions. 
When each partition at the parent RDD is used by multiple partitions of the child RDD or when
each partition from child produced or dependent on multiple parent RDD.

Slow as compare to narrow dependencies speed might be significantly affected as it might be 
required to shuffle data around different nodes when creating new partitions.

Might Require data shuffling over the cluster network or no data movement.

Functions such as groupByKey(), aggregateByKey(), aggregate(), join(), repartition() are some examples of wider transformations.

When working with Spark, it is always good or keeps in mind all operations or transformations Which 
might require data shuffling and hence slow down the process. 

Try to optimize and reduce the usage of wide dependencies as much as you can. 

____________________________________________________

Lazy Evaluation-
Lazy evaulation means that Spark will wait until the very last moment to execute the graph of
computation instructions. In Spark, instead of modifying the data immediately when you express
some operation, you build up a plan of transformations that you would like to apply to your
source data. By waiting until the last minute to execute the code, Spark compiles this plan from
your raw DataFrame transformations to a streamlined physical plan that will run as efficiently as
possible across the cluster. This provides some immense benifit as Spark can optimize the
entire data flow from end to end.

If we build a large Spark job but specify a filter at the end that only requires us to
fetch one row from our source data, the most efficient way to execute this is to access the single
record that we need. Spark will actually optimize this for us by pushing the filter down
automatically.


Actions-
Transformations allow us to build up our logical transformation plan. To trigger the computation,
we run an action. An action instructs Spark to compute a result from a series of transformations. 

Eg.

>>>> divisBy2.count()  # gives us the total number of records in the DataFrame

There are three kinds of actions:
1. Actions to view data in the console
2. Actions to collect data to native objects in the respective language
3. Actions to write to output data sources


Lets refer one eg.

Now, let’s sort our data according to the count column, which is an integer type.

Nothing happens to the data when we call sort because it’s just a transformation. However, we
can see that Spark is building up a plan for how it will execute this across the cluster by looking
at the explain plan. We can call "explain" on any DataFrame object to see the DataFrame’s
lineage (or how Spark will execute this query):

>>>> flightData2015.sort("count").explain()                       ### "explain" method is used to see the DataFrame’s lineage (or how Spark will execute this query)

O/P-
== Physical Plan ==
*Sort [count#195 ASC NULLS FIRST], true, 0
+- Exchange rangepartitioning(count#195 ASC NULLS FIRST, 200)
+- *FileScan csv [DEST_COUNTRY_NAME#193,ORIGIN_COUNTRY_NAME#194,count#195] ...

You can read "explain" plans from top to bottom, the
top being the end result, and the bottom being the source(s) of data. In this case, take a look at
the first keywords. You will see sort, exchange, and FileScan. That’s because the sort of our data
is actually a wide transformation because rows will need to be compared with one another.


The logical plan of transformations that we build up defines a lineage for the DataFrame so that
at any given point in time, Spark knows how to recompute any partition by performing all of the
operations it had before on the same input data. This sits at the heart of Spark’s programming
model—functional programming where the same inputs always result in the same outputs when
the transformations on that data stay constant.
We do not manipulate the physical data; instead, we configure physical execution characteristics
through things like the shuffle partitions parameter.


DATAFRAMES AND SQL

We worked through a simple transformation in the previous example, let’s now work through a
more complex one and follow along in both DataFrames and SQL. Spark can run the same
transformations, regardless of the language, in the exact same way. You can express your
business logic in SQL or DataFrames (either in R, Python, Scala, or Java) and Spark will
compile that logic down to an underlying plan (that you can see in the explain plan) before
actually executing your code. With Spark SQL, you can register any DataFrame as a table or
view (a temporary table) and query it using pure SQL. There is no performance difference
between writing SQL queries or writing DataFrame code, they both “compile” to the same
underlying plan that we specify in DataFrame code.

To make any DataFrame into a table or view, call this: 
>>>> flightData2015.createOrReplaceTempView("flight_data_2015") 


# in Python
sqlWay = spark.sql("""
SELECT DEST_COUNTRY_NAME, count(1)
FROM flight_data_2015
GROUP BY DEST_COUNTRY_NAME
""")


dataFrameWay = flightData2015\
.groupBy("DEST_COUNTRY_NAME")\
.count()
sqlWay.explain()
dataFrameWay.explain()


== Physical Plan ==
*HashAggregate(keys=[DEST_COUNTRY_NAME#182], functions=[count(1)])
+- Exchange hashpartitioning(DEST_COUNTRY_NAME#182, 5)
+- *HashAggregate(keys=[DEST_COUNTRY_NAME#182], functions=[partial_count(1)])
+- *FileScan csv [DEST_COUNTRY_NAME#182] ...
== Physical Plan ==
*HashAggregate(keys=[DEST_COUNTRY_NAME#182], functions=[count(1)])
+- Exchange hashpartitioning(DEST_COUNTRY_NAME#182, 5)
+- *HashAggregate(keys=[DEST_COUNTRY_NAME#182], functions=[partial_count(1)])
+- *FileScan csv [DEST_COUNTRY_NAME#182] ...


Let’s pull out some interesting statistics from our data. One thing to understand is that
DataFrames (and SQL) in Spark already have a huge number of manipulations available.

We will use the "max" function, to establish the maximum number of flights to and from any
given location. This just scans each value in the relevant column in the DataFrame and checks
whether it’s greater than the previous values that have been seen. This is a transformation,
because we are effectively filtering down to one row.

>>>> spark.sql("SELECT max(count) from flight_data_2015").take(1)

# in Python
>>>> from pyspark.sql.functions import max
>>>> flightData2015.select(max("count")).take(1)


Let’s perform something a bit more
complicated and find the top five destination countries in the data. This is our first multitransformation query,
so we’ll take it step by step. Let’s begin with a fairly straightforward SQL aggregation:

# in Python
>>>> maxSql = spark.sql("""
>>>> SELECT DEST_COUNTRY_NAME, sum(count) as destination_total
>>>> FROM flight_data_2015
>>>> GROUP BY DEST_COUNTRY_NAME
>>>> ORDER BY sum(count) DESC
>>>> LIMIT 5 """)

>>>> maxSql.show()
+-----------------+-----------------+
|DEST_COUNTRY_NAME|destination_total|
+-----------------+-----------------+
| United States| 411352|
| Canada| 8399|
| Mexico| 7140|
| United Kingdom| 2025|
| Japan| 1548|
+-----------------+-----------------+


Now, let’s move to the DataFrame syntax that is semantically similar but slightly different in
implementation and ordering

# in Python
>>>> from pyspark.sql.functions import desc
>>>> flightData2015\
>>>> .groupBy("DEST_COUNTRY_NAME")\
>>>> .sum("count")\
>>>> .withColumnRenamed("sum(count)", "destination_total")\
>>>> .sort(desc("destination_total"))\
>>>> .limit(5)\
>>>> .show()
+-----------------+-----------------+
|DEST_COUNTRY_NAME|destination_total|
+-----------------+-----------------+
| United States| 411352|
| Canada| 8399|
| Mexico| 7140|
| United Kingdom| 2025|
| Japan| 1548|
+-----------------+-----------------+


NOTE-- The execution plan is a directed acyclic graph (DAG) of transformations, each resulting in a new immutable DataFrame, on which we call an action to generate a result.



CHAPTER 3 -- A Tour of Spark’s Toolset

Building blocks of Apache Spark’s vast ecosystem of tools and libraries. 

Spark is composed of these primitives—the lower-level APIs (RDD, Distributed Variables) 
and the Structured APIs (Datasets, Dataframes, SQL)—and then a series of standard
libraries (Structural Streaming, Advanced Analytics, Libraries & Ecosystem) for additional functionality.
Spark’s libraries support a variety of different tasks, from graph analysis and machine learning to
streaming and integrations with a host of computing and storage systems.

This chapter covers the following:
1. Running production applications with spark-submit
2. Datasets: type-safe APIs for structured data
3. Structured Streaming
4. Machine learning and advanced analytics
5. Resilient Distributed Datasets (RDD): Spark’s low level APIs
6. SparkR
7. The third-party package ecosystem


1. Running production applications with spark-submit-


Spark makes it easy to develop and create big data programs. Spark also makes it easy to turn
your interactive exploration into production applications with a built-in command line tool "spark-submit".
It lets you send your application code to a cluster and launch it to execute there. 
Upon submission, the application will run until it exits (completes the task) or encounters an error.
You can do this with all of Spark’s support cluster managers including Standalone, Mesos, and YARN.

"spark-submit" offers several controls with which you can specify the resources your application
needs as well as how it should be run and its command-line arguments.

This sample application calculates the digits of pi to a certain level of estimation using Python. Here, we’ve
told spark-submit that we want to run on our local machine, which class and which JAR we
would like to run, and some command-line arguments for that class.

./bin/spark-submit \
--master local \
./examples/src/main/python/pi.py 10


2. Datasets: Type-Safe Structured APIs

The first API we’ll describe is a type-safe version of Spark’s structured API called "Datasets", for
writing statically typed code in Java and Scala. The "Dataset" API is NOT available in Python.

The Dataset API gives users the ability to assign a Java/Scala class to the records within a DataFrame
and manipulate it as a collection of typed objects, similar to a Java ArrayList or Scala Seq. 
The APIs available on Datasets are type-safe, meaning that you cannot accidentally view the objects 
in a Dataset as being of another class than the class you put in initially.

One great thing about Datasets is that you can use them only when you need or want to.


3. Structured Streaming-
Structured Streaming is a high-level API for stream processing that became production-ready in
Spark 2.2. With Structured Streaming, you can take the same operations that you perform in
batch mode using Spark’s structured APIs and run them in a streaming fashion. This can reduce
latency and allow for incremental processing. The best thing about Structured Streaming is that it
allows you to rapidly and quickly extract value out of streaming systems with virtually no code
changes. It also makes it easy to conceptualize because you can write your batch job as a way to
prototype it and then you can convert it to a streaming job. The way all of this works is by
incrementally processing that data.






CHAPTER 4- Structured API Overview

The Structured APIs are a tool for manipulating all sorts of data, from unstructured log files to
semi-structured CSV files and highly structured Parquet files. These APIs refer to three core types 
of distributed collection APIs:-

1. Datasets
2. DataFrames
3. SQL tables and views

Structured APIs apply to both batch and streaming computation. This means that when you work with the 
Structured APIs, it should be simple to migrate from batch to streaming (or vice versa) with little to no effort.


DataFrames and Datasets-

DataFrames and Datasets are (distributed) table-like collections with well-defined rows and
columns. Each column must have the same number of rows as all the other columns (although
you can use null to specify the absence of a value) and each column has type information that
must be consistent for every row in the collection. To Spark, DataFrames and Datasets represent
immutable, lazily evaluated plans that specify what operations to apply to data residing at a
location to generate some output. When we perform an action on a DataFrame, we instruct Spark
to perform the actual transformations and return the result. These represent plans of how to
manipulate rows and columns to compute the user’s desired result.

Schema-
A schema defines the column names and types of a DataFrame. You can define schemas
manually or read a schema from a data source (often called schema on read). Schemas consist of
types, meaning that you need a way of specifying what lies where.


Overview of Structured Spark Types-
Spark is effectively a programming language of its own. Internally, Spark uses an engine called
"Catalyst" that maintains its own type information through the planning and processing of work. In
doing so, this opens up a wide variety of execution optimizations that make significant
differences. Spark types map directly to the different language APIs that Spark maintains and
there exists a lookup table for each of these in Scala, Java, Python, SQL, and R. Even if we use
Spark’s Structured APIs from Python or R, the majority of our manipulations will operate strictly
on Spark types, not Python types. For example, the following code does not perform addition in
Scala or Python; it actually performs addition purely in Spark:

>>>> # in Python
>>>> df = spark.range(500).toDF("number")
>>>> df.select(df["number"] + 10)

This addition operation happens because Spark will convert an expression written in an input
language to Spark’s internal Catalyst representation of that same type information. It then will
operate on that internal representation.


DataFrames Versus Datasets-

In essence, within the Structured APIs, there are two more APIs, the “untyped” DataFrames and
the “typed” Datasets. To say that DataFrames are untyped is aslightly inaccurate; they have
types, but Spark maintains them completely and only checks whether those types line up to those
specified in the schema at runtime. Datasets, on the other hand, check whether types conform to
the specification at compile time. Datasets are only available to Java Virtual Machine (JVM)–
based languages (Scala and Java) and we specify types with case classes or Java beans.

To Spark (in Python or R), there is no such thing as a Dataset: everything is a
DataFrame and therefore we always operate on that optimized format.
When you’re using DataFrames, you’re taking advantage of Spark’s
optimized internal format. This format applies the same efficiency gains to all of Spark’s
language APIs.


Columns-

Columns represent a "simple" type like an integer or string, a "complex" type like an array or map, or
a "null" value. Spark tracks all of this type information for you and offers a variety of ways, with
which you can transform columns. For now, you can think about Spark Column types as columns in a table.

Rows-

A row is nothing more than a record of data. Each record in a DataFrame must be of type "Row", as
we can see when we collect the following DataFrames. We can create these rows manually from
SQL, from Resilient Distributed Datasets (RDDs), from data sources, or manually from scratch.

>>>> # in Python
>>>> spark.range(2).collect()		# This results in an array of "Row" objects.


Spark Types-

We mentioned earlier that Spark has a large number of internal type representations. We include
a handy reference table on the next several pages so that you can most easily reference what
type, in your specific language, lines up with the type in Spark.
Before getting to those tables, let’s talk about how we instantiate, or declare, a column to be of a
certain type.

Python types at times have certain requirements, which you can see listed below. To work with the
correct Python types, use the following:

>>>> from pyspark.sql.types import *
>>>> b = ByteType()


Overview of Structured API Execution-

This section will demonstrate how this code is actually executed across a cluster. This will help
you understand (and potentially debug) the process of writing and executing code on clusters, so
let’s walk through the execution of a single structured API query from user code to executed
code. Here’s an overview of the steps:
1. Write DataFrame/Dataset/SQL Code.
2. If valid code, Spark converts this to a "Logical Plan".
3. Spark transforms this "Logical Plan" to a "Physical Plan", checking for optimizations along
the way.
4. Spark then executes this "Physical Plan" (RDD manipulations) on the cluster.

To execute code, we must write code. This code is then submitted to Spark either through the
console or via a submitted job. This code then passes through the Catalyst Optimizer, which
decides how the code should be executed and lays out a plan for doing so before, finally, the
code is run and the result is returned to the user.

Rule. SQL/DataFrames/Datasets ----> Catalyst Optimizer -----> Physical Plan


A. Logical Planning- 
The first phase of execution is meant to take user code and convert it into a logical plan.

Rule. User Code --> Unresolved Logical Plan --(Analysis)--> Resolved Logical Plan --(Logical Optimization)--> Optimized Logical Plan
						  |
					          |
					       Catalog

This logical plan only represents a set of abstract transformations that do not refer to executors or
drivers, it’s purely to convert the user’s set of expressions into the most optimized version. It
does this by converting user code into an unresolved logical plan. This plan is unresolved
because although your code might be valid, the tables or columns that it refers to might or might
not exist. Spark uses the catalog, a repository of all table and DataFrame information, to resolve
columns and tables in the analyzer. The analyzer might reject the unresolved logical plan if the
required table or column name does not exist in the catalog. If the analyzer can resolve it, the
result is passed through the Catalyst Optimizer, a collection of rules that attempt to optimize the
logical plan by pushing down predicates or selections. Packages can extend the Catalyst to
include their own rules for domain-specific optimizations.


B. Physical Planning-
After successfully creating an optimized logical plan, Spark then begins the physical planning
process. 

Rule. Optimized Logical Plan --> Physical Plans --> Cost Model --> Best Physical Plan --> Executes on the cluster.

The physical plan, often called a Spark plan, specifies how the logical plan will execute
on the cluster by generating different physical execution strategies and comparing them through
a cost model. An example of the cost comparison might be choosing
how to perform a given join by looking at the physical attributes of a given table (how big the
table is or how big its partitions are).

Physical planning results in a series of RDDs and transformations. This result is why you might
have heard Spark referred to as a compiler—it takes queries in DataFrames, Datasets, and SQL
and compiles them into RDD transformations for you.


C. Execution-
Upon selecting a physical plan, Spark runs all of this code over RDDs, the lower-level
programming interface of Spark. Spark performs further
optimizations at runtime, generating native Java bytecode that can remove entire tasks or stages
during execution. Finally the result is returned to the user.


Deploy modes-
Whenever we submit spark application by using spark-submit, alongwith the spark-submit, we also need to specify "deploy mode".
There are 2 types of deploy mode- 

1. Client mode.
2. Cluster mode.

Let's say we submitted the spark application from client machine. There is one resource manager which will provide the resources 
in order to run the spark application. This resource manager will create container in worker nodes and in one of the container where the driver 
is running will be called as Application-Master container which will store your main program. So the execution of spark application actually starts from here i.e
from the worker node.

Client Mode - The spark driver component runs on the machine node where we submit the spark job. This mode support both interactive shells 
and the job submission commands.

Disadvantages are -
1. If the machine node fails then the entire job fails.
2. The performance of this mode is worst hence not preferred in production environment.

To overcome this issue, we use Cluster Mode for deployment in a production environment.
Cluster mode-
The spark job launches the driver component within the cluster as a part of the sub-process of Application Master.
We can use the spark-submit command for deployment and it does not support interactive shell mode.
There is a dedicated cluster manager such as YARN, Apache Mesos, Kubernetes, etc. for allocating the resources required for the job to run.
In case the program fails, the driver program is reinstatiated because the driver program are run in application master.
 

CHAPTER 5 - Basic Structured Operations


Definitionally, a DataFrame consists of a series of records (like rows in a table), that are of type
Row, and a number of columns (like columns in a spreadsheet) that represent a computation
expression that can be performed on each individual record in the Dataset. 
Schemas define the name as well as the type of data in each column. 
Partitioning of the DataFrame defines the layout of the DataFrame or Dataset’s physical distribution across the cluster. 
The partitioning scheme defines how that is allocated. You can set this to be based on values in a certain column
or nondeterministically.

Let’s create a DataFrame with which we can work:

>>>> # in Python
>>>> df = spark.read.format("json").load("/data/flight-data/json/2015-summary.json")

We discussed that a DataFame will have columns, and we use a schema to define them. Let’s
take a look at the schema on our current DataFrame:

>>>> df.printSchema()
Schemas tie everything together, so they’re worth belaboring. Let’s create a DataFrame with which we can work:

Schemas-

A schema defines the column names and types of a DataFrame. We can either let a data source
define the schema (called schema-on-read) or we can define it explicitly ourselves.

Let’s begin with a simple file, which we saw in Chapter 4, and let the semi-structured nature of
line-delimited JSON define the structure.

# in Python
>>>> spark.read.format("json").load("/data/flight-data/json/2015-summary.json").schema

Python returns the following:

StructType(List(StructField(DEST_COUNTRY_NAME,StringType,true),
StructField(ORIGIN_COUNTRY_NAME,StringType,true),
StructField(count,LongType,true)))

A schema is a StructType made up of a number of fields, StructFields, that have a name,
type, a Boolean flag which specifies whether that column can contain missing or null values,
and, finally, users can optionally specify associated metadata with that column. The metadata is a
way of storing information about this column.
Schemas can contain other StructTypes (Spark’s complex types).

The example that follows shows how to create and enforce a specific schema on a DataFrame-

>>>> from pyspark.sql.types import StructField, StructType, StringType, LongType
>>>> myManualSchema = StructType([
>>>> StructField("DEST_COUNTRY_NAME", StringType(), True),
>>>> StructField("ORIGIN_COUNTRY_NAME", StringType(), True),
>>>> StructField("count", LongType(), False, metadata={"hello":"world"}) ])
>>>> df = spark.read.format("json").schema(myManualSchema).load("/data/flight-data/json/2015-summary.json")

As discussed in Chapter 4, we cannot simply set types via the per-language types because Spark
maintains its own type information.


Columns and Expressions-

Columns in Spark are similar to columns in a spreadsheet, R dataframe, or pandas DataFrame.
You can select, manipulate, and remove columns from DataFrames and these operations are
represented as expressions.

To Spark, columns are logical constructions that simply represent a value computed on a perrecord basis by means of an expression. 
This means that to have a real value for a column, we
need to have a row; and to have a row, we need to have a DataFrame. You cannot manipulate an
individual column outside the context of a DataFrame; you must use Spark transformations
within a DataFrame to modify the contents of a column.

Columns-

There are a lot of different ways to construct and refer to columns but the two simplest ways are
by using the "col" or "column" functions. To use either of these functions, you pass in a column
name:

>>>> # in Python
>>>> from pyspark.sql.functions import col, column
>>>> col("someColumnName")
>>>> column("someColumnName")

As mentioned, this column might or might not exist in our DataFrames. Columns are not "resolved" 
until we compare the column names with those we are maintaining in the catalog. Column and table 
resolution happens in the analyzer phase, as discussed in Chapter 4.

Explicit column references-
If you need to refer to a specific DataFrame’s column, you can use the "col" method on the
specific DataFrame. This can be useful when you are performing a join and need to refer to a
specific column in one DataFrame that might share a name with another column in the joined
DataFrame. As an added benefit, Spark does not need to resolve
this column itself (during the analyzer phase) because we did that for Spark:

>>>> df.col("count")


Expressions-


An expression is a set of transformations on one or more values in a record in a DataFrame. Think of it like a
function that takes as input one or more column names, resolves them, and then potentially
applies more expressions to create a single value for each record in the dataset. Importantly, this
“single value” can actually be a complex type like a Map or Array.

An expression is created via the "expr" function (just a DataFrame column
reference). In the simplest case, expr("someCol") is equivalent to col("someCol")

Columns as Expressions-
Columns provide a subset of expression functionality. If you use col() and want to perform
transformations on that column, you must perform those on that column reference. When using
an expression, the "expr" function can actually parse transformations and column references from
a string and can subsequently be passed into further transformations. Let’s look at some
examples.

expr("someCol - 5") is the same transformation as performing col("someCol") - 5, or even
expr("someCol") - 5. That’s because Spark compiles these to a logical tree specifying the
order of operations.

Remember,
1. Columns are just expressions.
2. Columns and transformations of those columns compile to the same logical plan as
parsed expressions.

Let’s ground this with an example:
(((col("someCol") + 5) * 200) - 6) < col("otherCol")

Logical tree-
					<
					/\
				       /  \
				      /    \
				      -    OtherCol
				     /\
  				    /  \
   				   /    \
				  *      6
                                 /\
				/  \
			       /    \
                              +     200
			     /\
                            /  \
                           /    \
			SomeCol  5

>>>> # in Python
>>>> from pyspark.sql.functions import expr
>>>> expr("(((someCol + 5) * 200) - 6) < otherCol")

You can write your expressions as DataFrame code or as SQL
expressions and get the exact same performance characteristics. That’s because this SQL
expression and the previous DataFrame code compile to the same underlying logical tree prior to
execution.

Accessing a DataFrame’s columns-
Sometimes, you’ll need to see a DataFrame’s columns, which you can do by using something
like printSchema; however, if you want to programmatically access columns, you can use the
columns property to see all columns on a DataFrame:

>>>> spark.read.format("json").load("/data/flight-data/json/2015-summary.json").columns


Records and Rows-

In Spark, each row in a DataFrame is a single record. Spark represents this record as an object of
type "Row". Spark manipulates "Row" objects using column expressions in order to produce usable
values. "Row" objects internally represent arrays of bytes. The byte array interface is never shown
to users because we only use column expressions to manipulate them.

Let’s see a row by calling "first" on our DataFrame:
>>>> df.first()

Creating Rows-
You can create rows by manually instantiating a "Row" object with the values that belong in each
column. It’s important to note that only DataFrames have schemas. Rows themselves do not have
schemas. This means that if you create a Row manually, you must specify the values in the same
order as the schema of the DataFrame to which they might be appended.

>>> # in Python
>>> from pyspark.sql import Row
>>> myRow = Row("Hello", None, 1, False)

Accessing data in rows is equally as easy: you just specify the position that you would like.

>>>> # in Python
>>>> myRow[0]
>>>> myRow[2]


Dataframe Transformations-

Now, we will move onto manipulating DataFrames. When working with individual DataFrames 
there are some fundamental objectives. These break down into several core operations-

We can add rows or columns
We can remove rows or columns
We can transform a row into a column (or vice versa)
We can change the order of rows based on the values in columns (Sorting)


1. Creating Dataframes-
As we saw previously, we can create DataFrames from raw data source.

>>>> # in Python
>>>> df = spark.read.format("json").load("/data/flight-data/json/2015-summary.json")
>>>> df.createOrReplaceTempView("dfTable")

We can also create DataFrames on the fly by taking a set of rows and converting them to a
DataFrame.

>>>> # in Python
>>>> from pyspark.sql import Row
>>>> from pyspark.sql.types import StructField, StructType, StringType, LongType
>>>> myManualSchema = StructType([
>>>> StructField("some", StringType(), True),
>>>> StructField("col", StringType(), True),
>>>> StructField("names", LongType(), False)
>>>> ])
>>>> myRow = Row("Hello", None, 1)
>>>> myDf = spark.createDataFrame([myRow], myManualSchema)
>>>> myDf.show()

Giving an output of:
+-----+----+-----+
| some| col|names|
+-----+----+-----+
|Hello|null| 1|
+-----+----+-----+


Now, let’s take a look at most useful methods that
you’re going to be using: 
the "select" method when you’re working with columns or expressions,
and 
the "selectExpr" method when you’re working with expressions in strings. 

Naturally some transformations are not specified as methods on columns; 
therefore, there exists a group of
functions found in the "org.apache.spark.sql.functions" package.
With these three tools, you should be able to solve the vast majority of transformation challenges
that you might encounter in DataFrames.

select and selectExpr-
"select" and "selectExpr" allow you to do the DataFrame equivalent of SQL queries on a table of
data:

"
-- in SQL
SELECT * FROM dataFrameTable
SELECT columnName FROM dataFrameTable
SELECT columnName * 10, otherColumn, someOtherCol as c FROM dataFrameTable
"

In the simplest possible terms, you can use them to manipulate columns in your DataFrames.

Let’s walk through some examples on DataFrames to talk about some of the different ways of
approaching this problem. The easiest way is just to use the "select" method and pass in the
column names as strings with which you would like to work:

>>>> # in Python
>>>> df.select("DEST_COUNTRY_NAME").show(2)

"
-- in SQL
SELECT DEST_COUNTRY_NAME FROM dfTable LIMIT 2
"

Giving an output of:
+-----------------+
|DEST_COUNTRY_NAME|
+-----------------+
| United States|
| United States|
+-----------------+


You can select multiple columns by using the same style of query, just add more column name
strings to your select method call:

>>>> # in Python
>>>> df.select("DEST_COUNTRY_NAME", "ORIGIN_COUNTRY_NAME").show(2)

"
-- in SQL
SELECT DEST_COUNTRY_NAME, ORIGIN_COUNTRY_NAME FROM dfTable LIMIT 2
"

Giving an output of:
+-----------------+-------------------+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|
+-----------------+-------------------+
| United States| Romania|
| United States| Croatia|
+-----------------+-------------------+

You can refer to columns in a number of different ways; all you need to keep in mind is that you can use them interchangeably:

>>>> # in Python
>>>> from pyspark.sql.functions import expr, col, column
>>>> df.select(
>>>> expr("DEST_COUNTRY_NAME"),
>>>> col("DEST_COUNTRY_NAME"),
>>>> column("DEST_COUNTRY_NAME"))\
>>>> .show(2)

One common error is attempting to mix Column objects and strings.

For example, the following code will result in a compiler error:

>>>> df.select(col("DEST_COUNTRY_NAME"), "DEST_COUNTRY_NAME")

As we’ve seen thus far, "expr" is the most flexible reference that we can use. It can refer to a plain
column or a string manipulation of a column. To illustrate, let’s change the column name, and
then change it back by using the AS keyword and then the alias method on the column:

>>>> # in Python
>>>> df.select(expr("DEST_COUNTRY_NAME AS destination")).show(2)

"
-- in SQL
SELECT DEST_COUNTRY_NAME as destination FROM dfTable LIMIT 2
"
This changes the column name to “destination.” You can further manipulate the result of your
expression as another expression:


>>>> # in Python
>>>> df.select(expr("DEST_COUNTRY_NAME as destination").alias("DEST_COUNTRY_NAME")).show(2)

The preceding operation changes the column name back to its original name.

Because "select" followed by a series of "expr" is such a common pattern, Spark has a shorthand
for doing this efficiently: "selectExpr". This is probably the most convenient interface for
everyday use:

>>>> # in Python
>>>> df.selectExpr("DEST_COUNTRY_NAME as newColumnName", "DEST_COUNTRY_NAME").show(2)

This opens up the true power of Spark. We can treat "selectExpr" as a simple way to build up
complex expressions that create new DataFrames. In fact, we can add any valid non-aggregating
SQL statement, and as long as the columns resolve, it will be valid.

Here’s a simple example that
adds a new column "withinCountry" to our DataFrame that specifies whether the destination and
origin are the same:

>>>> # in Python
>>>> df.selectExpr(
>>>> "*", # all original columns
>>>> "(DEST_COUNTRY_NAME = ORIGIN_COUNTRY_NAME) as withinCountry").show(2)

"
-- in SQL
SELECT *, (DEST_COUNTRY_NAME = ORIGIN_COUNTRY_NAME) as withinCountry
FROM dfTable
LIMIT 2
"

Giving an output of:
+-----------------+-------------------+-----+-------------+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|withinCountry|
+-----------------+-------------------+-----+-------------+
| United States	  |     Romania       |  15 | false       |
| United States   |     Croatia	      |  1  | false       |


With select expression, we can also specify aggregations over the entire DataFrame by taking
advantage of the functions that we have. These look just like what we have been showing so far:

>>>> # in Python
>>>> df.selectExpr("avg(count)", "count(distinct(DEST_COUNTRY_NAME))").show(2)

"
-- in SQL
SELECT avg(count), count(distinct(DEST_COUNTRY_NAME)) FROM dfTable LIMIT 2
"
Giving an output of:
+-----------+---------------------------------+
| avg(count)|count(DISTINCT DEST_COUNTRY_NAME)|
+-----------+---------------------------------+
|1770.765625| 		132                   |
+-----------+---------------------------------+


Converting to Spark Types (Literals)-
Sometimes, we need to pass explicit values into Spark that are just a value (rather than a new
column). This might be a constant value or something we’ll need to compare to later on. The
way we do this is through literals. This is basically a translation from a given programming
language’s literal value to one that Spark understands. Literals are expressions and you can use
them in the same way:

>>>> # in Python
>>>> from pyspark.sql.functions import lit
>>>> df.select(expr("*"), lit(1).alias("One")).show(2)

In SQL, literals are just the specific value:

"
-- in SQL
SELECT *, 1 as One FROM dfTable LIMIT 2
"
Giving an output of:
+-----------------+-------------------+-----+---+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|One|
+-----------------+-------------------+-----+---+
| United States| Romania| 15| 1|
| United States| Croatia| 1| 1|
+-----------------+-------------------+-----+---+


Adding Columns-
There’s also a more formal way of adding a new column to a DataFrame, and that’s by using the
"withColumn" method on our DataFrame. For example,


>>>> # in Python
>>>> df.withColumn("numberOne", lit(1)).show(2)

"
-- in SQL
SELECT *, 1 as numberOne FROM dfTable LIMIT 2
"

Giving an output of:
+-----------------+-------------------+-----+---------+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|numberOne|
+-----------------+-------------------+-----+---------+
| United States| Romania| 15| 1|
| United States| Croatia| 1| 1|
+-----------------+-------------------+-----+---------+

In the next example,
we’ll set a Boolean flag for when the origin country is the same as the destination country:

>>>> # in Python
>>>> df.withColumn("withinCountry", expr("ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME")).show(2)

Interestingly, we can also rename a column this way.

Renaming Columns-
Although we can rename a column in the manner that we just described, another alternative is to
use the "withColumnRenamed" method. This will rename the column with the name of the string in
the first argument to the string in the second argument:

>>>> # in Python
>>>> df.withColumnRenamed("DEST_COUNTRY_NAME", "dest").columns


O/P: ... dest, ORIGIN_COUNTRY_NAME, count

Reserved Characters and Keywords-
One thing that you might come across is reserved characters like spaces or dashes in column
names. Handling these means escaping column names appropriately. In Spark, we do this by
using backtick (`) characters. Let’s use withColumn, which you just learned about to create a
column with reserved characters.

>>>> # in Python
>>>> dfWithLongColName = df.withColumn("This Long Column-Name", expr("ORIGIN_COUNTRY_NAME"))

We don’t need escape characters here because the first argument to "withColumn" is just a string
for the new column name.

In this example, however, we need to use backticks because we’re
referencing a column in an expression:

>>>> # in Python
>>>> dfWithLongColName.selectExpr(
>>>> "`This Long Column-Name`",
>>>> "`This Long Column-Name` as `new col`").show(2)
>>>> dfWithLongColName.createOrReplaceTempView("dfTableLong")

"
-- in SQL
SELECT `This Long Column-Name`, `This Long Column-Name` as `new col`
FROM dfTableLong LIMIT 2
"

Case Sensitivity-
By default Spark is case insensitive; however, you can make Spark case sensitive by setting the
configuration:

"
-- in SQL
set spark.sql.caseSensitive true
"

Removing Columns-
Let’s take a look at how we can remove columns from
DataFrames. You likely already noticed that we can do this by using select. However, there is
also a dedicated method called "drop":

>>>> df.drop("ORIGIN_COUNTRY_NAME").columns

We can drop multiple columns by passing in multiple columns as arguments:

>>>> dfWithLongColName.drop("ORIGIN_COUNTRY_NAME", "DEST_COUNTRY_NAME")


Changing a Column’s Type (cast)-
Sometimes, we might need to convert from one type to another; for example, if we have a set of
StringType that should be integers. We can convert columns from one type to another by
casting the column from one type to another. For instance, let’s convert our count column from
an integer to a type Long:

>>>> df.withColumn("count2", col("count").cast("long"))

"
-- in SQL
SELECT *, cast(count as long) AS count2 FROM dfTable
"

Filtering Rows-
To filter rows, we create an expression that evaluates to true or false. You then filter out the rows
with an expression that is equal to false. The most common way to do this with DataFrames is to
create either an expression as a String or build an expression by using a set of column
manipulations. 
There are two methods to perform this operation: you can use "where" or "filter"
and they both will perform the same operation and accept the same argument types when used
with DataFrames.

>>>> df.filter(col("count") < 2).show(2)     ### using "filter" method

>>>> df.where("count < 2").show(2)	     ### using "where" method

"
-- in SQL
SELECT * FROM dfTable WHERE count < 2 LIMIT 2
"

Giving an output of:
+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
| United States| Croatia| 1|
| United States| Singapore| 1|
+-----------------+-------------------+-----+


You might want to put multiple filters into the same expression. Although this is
possible, it is not always useful, because Spark automatically performs all filtering operations at
the same time regardless of the filter ordering. This means that if you want to specify multiple
AND filters, just chain them sequentially and let Spark handle the rest:

>>>> # in Python
>>>> df.where(col("count") < 2).where(col("ORIGIN_COUNTRY_NAME") != "Croatia").show(2)

"
-- in SQL
SELECT * FROM dfTable WHERE count < 2 AND ORIGIN_COUNTRY_NAME != "Croatia"
LIMIT 2
"

Giving an output of:
+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
| United States| Singapore| 1|
| Moldova| United States| 1|
+-----------------+-------------------+-----+


Getting Unique Rows-
A very common use case is to extract the unique or distinct values in a DataFrame. These values
can be in one or more columns. The way we do this is by using the distinct method on a
DataFrame, which allows us to deduplicate any rows that are in that DataFrame. 

>>>> # in Python
>>>> df.select("ORIGIN_COUNTRY_NAME", "DEST_COUNTRY_NAME").distinct().count()

"
-- in SQL
SELECT COUNT(DISTINCT(ORIGIN_COUNTRY_NAME, DEST_COUNTRY_NAME)) FROM dfTable
"

Results in 256.

>>>> # in Python
>>>> df.select("ORIGIN_COUNTRY_NAME").distinct().count()


"
-- in SQL
SELECT COUNT(DISTINCT ORIGIN_COUNTRY_NAME) FROM dfTable
"

Results in 125.


Random Samples-
Sometimes, you might just want to sample some random records from your DataFrame. You can
do this by using the "sample" method on a DataFrame and whether you’d like to sample with or without
replacement:

>>>> # in Python
>>>> seed = 5
>>>> withReplacement = False
>>>> fraction = 0.5
>>>> df.sample(withReplacement, fraction, seed).count()

Giving an output of 126.


Random Splits-
Random splits can be helpful when you need to break up your DataFrame into a random “splits”
of the original DataFrame. This is often used with machine learning algorithms to create training,
validation, and test sets. Because this method is designed to be randomized, we will also
specify a seed (just replace seed with a number of your choosing in the code block).

>>>> # in Python
>>>> dataFrames = df.randomSplit([0.25, 0.75], seed)
>>>> dataFrames[0].count() > dataFrames[1].count()	 # False


Concatenating and Appending Rows (Union)-
As you learned in the previous section, DataFrames are immutable. This means users cannot
append to DataFrames because that would be changing it. To append to a DataFrame, you must
union the original DataFrame along with the new DataFrame. This just concatenates the two
DataFramess. To union two DataFrames, you must be sure that they have the same schema and
number of columns; otherwise, the union will fail.

>>>> # in Python
>>>> from pyspark.sql import Row
>>>> schema = df.schema
>>>> newRows = [
>>>> Row("New Country", "Other Country", 5L),
>>>> Row("New Country 2", "Other Country 3", 1L)
>>>> ]
>>>> parallelizedRows = spark.sparkContext.parallelize(newRows)
>>>> newDF = spark.createDataFrame(parallelizedRows, schema)

>>>> # in Python
>>>> df.union(newDF)\
>>>> .where("count = 1")\
>>>> .where(col("ORIGIN_COUNTRY_NAME") != "United States")\
>>>> .show()

Giving the output of:
+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
| United States| Croatia| 1|
...
| United States| Namibia| 1|
| New Country 2| Other Country 3| 1|
+-----------------+-------------------+-----+



Sorting Rows-
When we sort the values in a DataFrame, we always want to sort with either the largest or
smallest values at the top of a DataFrame. There are two equivalent operations to do this "sort"
and "orderBy" that work the exact same way. They accept both column expressions and strings as
well as multiple columns. The default is to sort in ascending order:

# in Python
df.sort("count").show(5)
df.orderBy("count", "DEST_COUNTRY_NAME").show(5)
df.orderBy(col("count"), col("DEST_COUNTRY_NAME")).show(5)

To more explicitly specify sort direction, you need to use the asc and desc functions if operating
on a column. These allow you to specify the order in which a given column should be sorted:

# in Python
from pyspark.sql.functions import desc, asc
df.orderBy(expr("count desc")).show(2)
df.orderBy(col("count").desc(), col("DEST_COUNTRY_NAME").asc()).show(2)

-- in SQL
SELECT * FROM dfTable ORDER BY count DESC, DEST_COUNTRY_NAME ASC LIMIT 2

An advanced tip is to use "asc_nulls_first", "desc_nulls_first", "asc_nulls_last", or
"desc_nulls_last" to specify where you would like your null values to appear in an ordered
DataFrame.

For optimization purposes, it’s sometimes advisable to sort within each partition before another
set of transformations. You can use the sortWithinPartitions method to do this:

# in Python
spark.read.format("json").load("/data/flight-data/json/*-summary.json")\
.sortWithinPartitions("count")


Limit-
Oftentimes, you might want to restrict what you extract from a DataFrame; You can do this by using the limit method:

# in Python
df.limit(5).show()

-- in SQL
SELECT * FROM dfTable LIMIT 6

# in Python
df.orderBy(expr("count desc")).limit(6).show()

-- in SQL
SELECT * FROM dfTable ORDER BY count desc LIMIT 6



CHAPTER 6- Working with Different Types of Data-

This chapter covers building expressions, which are the bread and butter of Spark’s structured operations.

We also review working with a variety of different kinds of data, including the following:
Booleans
Numbers
Strings
Dates and timestamps
Handling null
Complex types
User-defined functions

Where to Look for APIs-

1. DataFrame (Dataset) Methods
This is actually a bit of a trick because a DataFrame is just a Dataset of Row types, so you’ll
actually end up looking at the Dataset methods. Refer ("https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.Dataset")

Dataset submodules like "DataFrameStatFunctions" and "DataFrameNaFunctions" have more
methods that solve specific sets of problems. 
"DataFrameStatFunctions" holds a variety of statistically related functions, 
"DataFrameNaFunctions" refers to functions that are relevant when working with null data.

2. Column Methods
It holds a variety of general columnrelated methods like "alias" or "contain". Refer ("https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.Column")

"org.apache.spark.sql.functions" contains a variety of functions for a range of different data
types. Often, you’ll see the entire package imported because they are used so frequently. Refer ("https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.functions$")

Let’s read in the DataFrame that we’ll be using for this analysis:

>>>> # in Python
>>>> df = spark.read.format("csv")\
>>>> .option("header", "true")\
>>>> .option("inferSchema", "true")\
>>>> .load("/data/retail-data/by-day/2010-12-01.csv")
>>>> df.printSchema()
>>>> df.createOrReplaceTempView("dfTable")

Here’s the result of the schema and a small sample of the data:
"
root
|-- InvoiceNo: string (nullable = true)
|-- StockCode: string (nullable = true)
|-- Description: string (nullable = true)
|-- Quantity: integer (nullable = true)
|-- InvoiceDate: timestamp (nullable = true)
|-- UnitPrice: double (nullable = true)
|-- CustomerID: double (nullable = true)
|-- Country: string (nullable = true)
"

"
+---------+---------+--------------------+--------+-------------------+----...
|InvoiceNo|StockCode| Description|Quantity| InvoiceDate|Unit...
+---------+---------+--------------------+--------+-------------------+----...
| 536365| 85123A|WHITE HANGING HEA...| 6|2010-12-01 08:26:00| ...
| 536365| 71053| WHITE METAL LANTERN| 6|2010-12-01 08:26:00| ...
...
| 536367| 21755|LOVE BUILDING BLO...| 3|2010-12-01 08:34:00| ...
| 536367| 21777|RECIPE BOX WITH M...| 4|2010-12-01 08:34:00| ...
+---------+---------+--------------------+--------+-------------------+----...
"

Converting to Spark Types-

One thing you’ll see us do throughout this chapter is convert native types to Spark types. We do
this by using the "lit" function. This function converts a type in another language to its correspnding Spark representation.

>>> # in Python
>>>> from pyspark.sql.functions import lit
>>>> df.select(lit(5), lit("five"), lit(5.0))

There’s no equivalent function necessary in SQL, so we can use the values directly:

"
-- in SQL
SELECT 5, "five", 5.0
"

Working with Booleans-

Booleans are essential when it comes to data analysis because they are the foundation for all
filtering. 
Boolean statements consist of four elements: "and", "or", "true", and "false". We can specify equality as well as
less-than or greater-than:

>>>> # in Python
>>>> from pyspark.sql.functions import col
>>>> df.where(col("InvoiceNo") != 536365)\
>>>> .select("InvoiceNo", "Description")\
>>>> .show(5, False)

"
+---------+-----------------------------+
|InvoiceNo|Description |
+---------+-----------------------------+
|536366 |HAND WARMER UNION JACK |
...
|536367 |POPPY'S PLAYHOUSE KITCHEN |
+---------+-----------------------------+
"

Another option—and probably the cleanest—is to specify the predicate as an expression in a
string. This is valid for Python. Note that this also gives you access to another way of
expressing “does not equal”:

>>>> df.where("InvoiceNo = 536365").show(5, false)
>>>> df.where("InvoiceNo <> 536365").show(5, false)

In Spark, you should always chain together and filters as a sequential filter.
The reason is that even if Boolean statements are expressed serially (one after the other),
Spark will flatten all of these filters into one statement and perform the filter at the same time,
creating the "and" statement for us.
Although you can specify your statements explicitly by using
"and" if you like, they’re often easier to understand and to read if you specify them serially. "or"
statements need to be specified in the same statement:

>>>> # in Python
>>>> from pyspark.sql.functions import instr
>>>> priceFilter = col("UnitPrice") > 600
>>>> descripFilter = instr(df.Description, "POSTAGE") >= 1
>>>> df.where(df.StockCode.isin("DOT")).where(priceFilter | descripFilter).show()

"
-- in SQL
SELECT * FROM dfTable WHERE StockCode in ("DOT") AND(UnitPrice > 600 OR
instr(Description, "POSTAGE") >= 1)
"

+---------+---------+--------------+--------+-------------------+---------+...
|InvoiceNo|StockCode| Description|Quantity| InvoiceDate|UnitPrice|...
+---------+---------+--------------+--------+-------------------+---------+...
| 536544| DOT|DOTCOM POSTAGE| 1|2010-12-01 14:32:00| 569.77|...
| 536592| DOT|DOTCOM POSTAGE| 1|2010-12-01 17:06:00| 607.49|...
+---------+---------+--------------+--------+-------------------+---------+...


Boolean expressions are not just reserved to filters. To filter a DataFrame, you can also just
specify a Boolean column:

>>>> # in Python
>>>> from pyspark.sql.functions import instr
>>>> DOTCodeFilter = col("StockCode") == "DOT"
>>>> priceFilter = col("UnitPrice") > 600
>>>> descripFilter = instr(col("Description"), "POSTAGE") >= 1
>>>> df.withColumn("isExpensive", DOTCodeFilter & (priceFilter | descripFilter))\
>>>> .where("isExpensive")\
>>>> .select("unitPrice", "isExpensive").show(5)

"
-- in SQL
SELECT UnitPrice, (StockCode = 'DOT' AND
(UnitPrice > 600 OR instr(Description, "POSTAGE") >= 1)) as isExpensive
FROM dfTable
WHERE (StockCode = 'DOT' AND
(UnitPrice > 600 OR instr(Description, "POSTAGE") >= 1))
"

Here, we did not need to specify our filter as an expression and we used a column name without any extra work.

If you’re coming from a SQL background, all of these statements should seem quite familiar.
Indeed, all of them can be expressed as a "where" clause. In fact, it’s often easier to just express
filters as SQL statements than using the programmatic DataFrame interface and Spark SQL
allows us to do this without paying any performance penalty. Like this-

>>>> # in Python
>>>> from pyspark.sql.functions import expr
>>>> df.withColumn("isExpensive", expr("NOT UnitPrice <= 250"))\
>>>> .where("isExpensive")\
>>>> .select("Description", "UnitPrice").show(5)


Working with Numbers-
To fabricate a contrived example, let’s imagine that we found out that we mis-recorded the
quantity in our retail dataset and the true quantity is equal to (the current quantity * the unit
price) + 5. This will introduce our first numerical function as well as the pow function that raises
a column to the expressed power:

>>>> # in Python
>>>> from pyspark.sql.functions import expr, pow
>>>> fabricatedQuantity = pow(col("Quantity") * col("UnitPrice"), 2) + 5
>>>> df.select(expr("CustomerId"), fabricatedQuantity.alias("realQuantity")).show(2)

+----------+------------------+
2
|CustomerId| realQuantity|
+----------+------------------+
| 17850.0|239.08999999999997|
| 17850.0| 418.7156|
+----------+------------------+

In fact, we can do all of this as a SQL
expression, as well:

>>>> # in Python
>>>> df.selectExpr(
>>>> "CustomerId",
>>>> "(POWER((Quantity * UnitPrice), 2.0) + 5) as realQuantity").show(2)

"
-- in SQL
SELECT customerId, (POWER((Quantity * UnitPrice), 2.0) + 5) as realQuantity
FROM dfTable
"

Another common numerical task is rounding. By default, the "round" function rounds up if you’re exactly in between two numbers. You can
round down by using the "bround":

>>>> # in Python
>>>> from pyspark.sql.functions import lit, round, bround
>>>> df.select(round(lit("2.5")), bround(lit("2.5"))).show(2)

"
-- in SQL
SELECT round(2.5), bround(2.5)
"

+-------------+--------------+
|round(2.5, 0)|bround(2.5, 0)|
+-------------+--------------+
| 3.0| 2.0|
| 3.0| 2.0|
+-------------+--------------+

Another numerical task is to compute the correlation of two columns. For example, we can see
the Pearson correlation coefficient for two columns to see if cheaper things are typically bought
in greater quantities.

>>>> # in Python
>>>> from pyspark.sql.functions import corr
>>>> df.stat.corr("Quantity", "UnitPrice")		## Method 1
>>>> df.select(corr("Quantity", "UnitPrice")).show()		## Method 2

"
-- in SQL
SELECT corr(Quantity, UnitPrice) FROM dfTable
+-------------------------+
|corr(Quantity, UnitPrice)|
+-------------------------+
| -0.04112314436835551|
+-------------------------+
"

Another common task is to compute summary statistics for a column or set of columns. We can
use the "describe" method to achieve exactly this.

>>>> # in Python
>>>> df.describe().show()

+-------+------------------+------------------+------------------+
|summary| Quantity| UnitPrice| CustomerID|

+-------+------------------+------------------+------------------+
| count| 3108| 3108| 1968|
| mean| 8.627413127413128| 4.151946589446603|15661.388719512195|
| stddev|26.371821677029203|15.638659854603892|1854.4496996893627|
| min| -24| 0.0| 12431.0|
| max| 600| 607.49| 18229.0|
+-------+------------------+------------------+------------------+

If you need these exact numbers, you can also perform this as an aggregation yourself by
importing the functions and applying them to the columns that you need:

>>>> # in Python
>>>> from pyspark.sql.functions import count, mean, stddev_pop, min, max

There are a number of statistical functions available in the StatFunctions Package (accessible
using "stat" as we see in the code block below). These are DataFrame methods that you can use
to calculate a variety of different things. For instance, you can calculate either exact or
approximate quantiles of your data using the "approxQuantile" method:

>>>> # in Python
>>>> colName = "UnitPrice"
>>>> quantileProbs = [0.5]
>>>> relError = 0.05
>>>> df.stat.approxQuantile("UnitPrice", quantileProbs, relError) # 2.51

You also can use this to see a cross-tabulation or frequent item pairs (be careful, this output will
be large and is omitted for this reason):

>>>> # in Python
>>>> df.stat.crosstab("StockCode", "Quantity").show()

>>>> # in Python
>>>> df.stat.freqItems(["StockCode", "Quantity"]).show()

we can also add a unique ID to each row by using the function
"monotonically_increasing_id". This function generates a unique value for each row, starting
with 0:


>>>> # in Python
>>>> from pyspark.sql.functions import monotonically_increasing_id
>>>> df.select(monotonically_increasing_id()).show(2)


Working with Strings-

String manipulation shows up in nearly every data flow, and it’s worth explaining what you can
do with strings. You might be manipulating log files performing regular expression extraction or
substitution, or checking for simple string existence, or making all strings uppercase or
lowercase.

The "initcap" function will capitalize every word in a given string when that word is separated from another by a space.

>>>> # in Python
>>>> from pyspark.sql.functions import initcap
>>>> df.select(initcap(col("Description"))).show()

"
-- in SQL
SELECT initcap(Description) FROM dfTable
"

+----------------------------------+
|initcap(Description) |
+----------------------------------+
|White Hanging Heart T-light Holder|
|White Metal Lantern |
+----------------------------------+

As just mentioned, you can cast strings in uppercase and lowercase, as well:

>>>> # in Python
>>>> from pyspark.sql.functions import lower, upper
>>>> df.select(col("Description"),lower(col("Description")),upper(lower(col("Description")))).show(2)

"
-- in SQL
SELECT Description, lower(Description), Upper(lower(Description)) FROM dfTable
"

+--------------------+--------------------+-------------------------+
| Description| lower(Description)|upper(lower(Description))|
+--------------------+--------------------+-------------------------+
|WHITE HANGING HEA...|white hanging hea...| WHITE HANGING HEA...|
| WHITE METAL LANTERN| white metal lantern| WHITE METAL LANTERN|
+--------------------+--------------------+-------------------------+

Another trivial task is adding or removing spaces around a string. 
You can do this by using "lpad", "ltrim", "rpad" and "rtrim", "trim"

>>>> # in Python
>>>> from pyspark.sql.functions import lit, ltrim, rtrim, rpad, lpad, trim
>>>> df.select(
>>>> ltrim(lit(" HELLO ")).alias("ltrim"),
>>>> rtrim(lit(" HELLO ")).alias("rtrim"),
>>>> trim(lit(" HELLO ")).alias("trim"),
>>>> lpad(lit("HELLO"), 3, " ").alias("lp"),
>>>> rpad(lit("HELLO"), 10, " ").alias("rp")).show(2)

-- in SQL
SELECT
ltrim(' HELLLOOOO '),
rtrim(' HELLLOOOO '),
trim(' HELLLOOOO '),
lpad('HELLOOOO ', 3, ' '),
rpad('HELLOOOO ', 10, ' ')
FROM dfTable

+---------+---------+-----+---+----------+
| ltrim| rtrim| trim| lp| rp|
+---------+---------+-----+---+----------+
|HELLO | HELLO|HELLO| HE|HELLO |
|HELLO | HELLO|HELLO| HE|HELLO |
+---------+---------+-----+---+----------+

Note that if "lpad" or "rpad" takes a number less than the length of the string, it will always remove
values from the right side of the string.


Regular Expressions-

Regular Expressions are most frequently performed tasks for searching the existence of one string
in another or replacing all mentions of a string with another value. It gives
the user an ability to specify a set of rules to use to either extract values from a string or replace
them with some other values.

There are two key functions in Spark that you’ll need in
order to perform regular expression tasks: "regexp_extract" and "regexp_replace". These functions extract values and replace values, respectively.

Let’s explore how to use the "regexp_replace" function to replace substitute color names in our
description column:

>>>> # in Python
>>>> from pyspark.sql.functions import regexp_replace
>>>> regex_string = "BLACK|WHITE|RED|GREEN|BLUE"
>>>> df.select(
>>>> regexp_replace(col("Description"), regex_string, "COLOR").alias("color_clean"),col("Description")).show(2)

-- in SQL
SELECT
regexp_replace(Description, 'BLACK|WHITE|RED|GREEN|BLUE', 'COLOR') as
color_clean, Description
FROM dfTable

+--------------------+--------------------+
| color_clean| Description|
+--------------------+--------------------+
|COLOR HANGING HEA...|WHITE HANGING HEA...|
| COLOR METAL LANTERN| WHITE METAL LANTERN|
+--------------------+--------------------+

Another task might be to replace given characters with other characters. Building this as a
regular expression could be tedious, so Spark also provides the "translate" function to replace these
values. This is done at the character level and will replace all instances of a character with the
indexed character in the replacement string:

>>>> # in Python
>>>> from pyspark.sql.functions import translate
>>>> df.select(translate(col("Description"), "LEET", "1337"),col("Description")).show(2)

-- in SQL
SELECT translate(Description, 'LEET', '1337'), Description FROM dfTable

+----------------------------------+--------------------+
|translate(Description, LEET, 1337)| Description|
+----------------------------------+--------------------+
| WHI73 HANGING H3A...|WHITE HANGING HEA...|
| WHI73 M37A1 1AN73RN| WHITE METAL LANTERN|
+----------------------------------+--------------------+

We can also perform something similar, like pulling out the first mentioned color:

>>>> # in Python
>>>> from pyspark.sql.functions import regexp_extract
>>>> extract_str = "(BLACK|WHITE|RED|GREEN|BLUE)"
>>>> df.select(regexp_extract(col("Description"), extract_str, 1).alias("color_clean"),col("Description")).show(2)

-- in SQL
SELECT regexp_extract(Description, '(BLACK|WHITE|RED|GREEN|BLUE)', 1),
Description
FROM dfTable

+-------------+--------------------+
| color_clean| Description|
+-------------+--------------------+
| WHITE|WHITE HANGING HEA...|
| WHITE| WHITE METAL LANTERN|
+-------------+--------------------+

Sometimes, rather than extracting values, we simply want to check for their existence. We can do
this with the "instr" method on each column. This will return a Boolean declaring whether the
value you specify is in the column’s string:

In Python and SQL, we can do it like:

>>>> # in Python
>>>> from pyspark.sql.functions import instr
>>>> containsBlack = instr(col("Description"), "BLACK") >= 1
>>>> containsWhite = instr(col("Description"), "WHITE") >= 1
>>>> df.withColumn("hasSimpleColor", containsBlack | containsWhite)\
>>>> .where("hasSimpleColor")\
>>>> .select("Description").show(3, False)

-- in SQL
SELECT Description FROM dfTable
WHERE instr(Description, 'BLACK') >= 1 OR instr(Description, 'WHITE') >= 1
+----------------------------------+
|Description |
+----------------------------------+
|WHITE HANGING HEART T-LIGHT HOLDER|
|WHITE METAL LANTERN |
|RED WOOLLY HOTTIE WHITE HEART. |
+----------------------------------+


Let’s work through this in a more rigorous way and take advantage of Spark’s ability to accept a dynamic number of arguments. 
When we convert a list of values into a set of arguments and pass them into a function, we use a language feature called "varargs". 
Using this feature, we can
effectively unravel an array of arbitrary length and pass it as arguments to a function. This,
coupled with "select" makes it possible for us to create arbitrary numbers of columns
dynamically.

We can also do this quite easily in Python. In this case, we’re going to use a different function,
"locate", that returns the integer location (1 based location). We then convert that to a Boolean
before using it as the same basic feature:

>>>> # in Python
>>>> from pyspark.sql.functions import expr, locate
>>>> simpleColors = ["black", "white", "red", "green", "blue"]
>>>> def color_locator(column, color_string):
>>>> 	return locate(color_string.upper(), column).cast("boolean").alias("is_" + c)
>>>>
>>>> selectedColumns = [color_locator(df.Description, c) for c in simpleColors]
>>>> selectedColumns.append(expr("*")) # has to a be Column type
>>>> df.select(*selectedColumns).where(expr("is_white OR is_red")).select("Description").show(3, False)

This simple feature can often help you programmatically generate columns or Boolean filters in a
way that is simple to understand and extend. We could extend this to calculating the smallest
common denominator for a given input value, or whether a number is a prime.


Working with Dates and Timestamps-

Spark, as we saw with our current dataset, will make a best effort to
correctly identify column types, including dates and timestamps when we enable "inferSchema".
Working with dates and timestamps closely relates to working with strings
because we often store our timestamps or dates as strings and convert them into date types at
runtime. This is less common when working with databases and structured data but much more
common when we are working with text and CSV files.

Although Spark will do read dates or times on a best-effort basis. However, sometimes there will
be no getting around working with strangely formatted dates and times.

The key to understanding the transformations that you are going to need to apply is to ensure that you know
exactly what type and format you have at each given step of the way. Spark’s "TimestampType" class supports only second-level precision, which means that if
you’re going to be working with milliseconds or microseconds, you’ll need to work around this
problem by potentially operating on them as "longs". Any more precision when coercing to a
TimestampType will be removed.

Let’s begin with the basics and get the current date and the current timestamps:

>>>> # in Python
>>>> from pyspark.sql.functions import current_date, current_timestamp
>>>> dateDF = spark.range(10)\
>>>> .withColumn("today", current_date())\
>>>> .withColumn("now", current_timestamp())
>>>> dateDF.createOrReplaceTempView("dateTable")
>>>> dateDF.printSchema()

root
|-- id: long (nullable = false)
|-- today: date (nullable = false)
|-- now: timestamp (nullable = false)


Now that we have a simple DataFrame to work with, let’s add and subtract five days from today.
These functions take a column and then the number of days to either add or subtract as the
arguments:

>>>> # in Python
>>>> from pyspark.sql.functions import date_add, date_sub
>>>> dateDF.select(date_sub(col("today"), 5), date_add(col("today"), 5)).show(1)

-- in SQL
SELECT date_sub(today, 5), date_add(today, 5) FROM dateTable

+------------------+------------------+
|date_sub(today, 5)|date_add(today, 5)|
+------------------+------------------+
| 2017-06-12| 2017-06-22|
+------------------+------------------+

"datediff" function that will return the number of days in between two dates. "months_between" gives you the number of months between two dates.


# in Python
from pyspark.sql.functions import datediff, months_between, to_date
dateDF.withColumn("week_ago", date_sub(col("today"), 7))\
.select(datediff(col("week_ago"), col("today"))).show(1)
dateDF.select(
to_date(lit("2016-01-01")).alias("start"),
to_date(lit("2017-05-22")).alias("end"))\
.select(months_between(col("start"), col("end"))).show(1)
-- in SQL
SELECT to_date('2016-01-01'), months_between('2016-01-01', '2017-01-01'),
datediff('2016-01-01', '2017-01-01')
FROM dateTable
+-------------------------+
|datediff(week_ago, today)|
+-------------------------+
| -7|
+-------------------------+
+--------------------------+
|months_between(start, end)|
+--------------------------+
| -16.67741935|


The "to_date" function allows you to convert a string to a date.

>>>> # in Python
>>>> from pyspark.sql.functions import to_date, lit
>>>> spark.range(5).withColumn("date", lit("2017-01-01")).select(to_date(col("date"))).show(1)

Spark will not throw an error if it cannot parse the date; rather, it will just return null.
For this, let’s take a look at the date format that has switched from year-
month-day to year-day-month. Spark will fail to parse this date and silently return "null" instead:

>>>> dateDF.select(to_date(lit("2016-20-12")),to_date(lit("2017-12-11"))).show(1)

+-------------------+-------------------+
|to_date(2016-20-12)|to_date(2017-12-11)|
+-------------------+-------------------+
| null| 2017-12-11|
+-------------------+-------------------+

We find this to be an especially tricky situation for bugs because some dates might match the
correct format, whereas others do not. Let’s fix this pipeline, step by step, and come up with a robust way to avoid these issues entirely.
The first step is to remember that we need to specify our date format according to the Java
SimpleDateFormat standard [https://docs.oracle.com/javase/8/docs/api/java/text/SimpleDateFormat.html].

We will use two functions to fix this: "to_date" and "to_timestamp".

>>>> # in Python
>>>> from pyspark.sql.functions import to_date
>>>> dateFormat = "yyyy-dd-MM"
>>>> cleanDateDF = spark.range(1).select(
>>>> to_date(lit("2017-12-11"), dateFormat).alias("date"),
>>>> to_date(lit("2017-20-12"), dateFormat).alias("date2"))
>>>> cleanDateDF.createOrReplaceTempView("dateTable2")

-- in SQL
SELECT to_date(date, 'yyyy-dd-MM'), to_date(date2, 'yyyy-dd-MM'), to_date(date)
FROM dateTable2

+----------+----------+
| date| date2|
+----------+----------+
|2017-11-12|2017-12-20|
+----------+----------+

Now let’s use an example of "to_timestamp", which always requires a format to be specified:

>>>> # in Python
>>>> from pyspark.sql.functions import to_timestamp
>>>> dateFormat = "yyyy-dd-MM"
>>>> cleanDateDF.select(to_timestamp(col("date"), dateFormat)).show()

-- in SQL
SELECT to_timestamp(date, 'yyyy-dd-MM'), to_timestamp(date2, 'yyyy-dd-MM')
FROM dateTable2

+----------------------------------+
|to_timestamp(`date`, 'yyyy-dd-MM')|
+----------------------------------+
| 2017-11-12 00:00:00|
+----------------------------------+
 
After we have our date or timestamp in the correct format and type, comparing between them is
actually quite easy. We just need to be sure to either use a date/timestamp type or specify our
string according to the right format of yyyy-MM-dd if we’re comparing a date:

>>>> cleanDateDF.filter(col("date2") > lit("2017-12-12")).show()

One minor point is that we can also set this as a string, which Spark parses to a literal:

>>>> cleanDateDF.filter(col("date2") > "'2017-12-12'").show()


Working with Nulls in Data-

You should always use nulls to represent missing or empty data in your
DataFrames. Spark can optimize working with null values more than it can if you use empty
strings or other values.

Use the ".na" subpackage on a DataFrame for interacting with null values.

There are two things you can do with null values: you can explicitly drop nulls or you can fill
them with a value (globally or on a per-column basis). Let’s experiment with each of these now.


Coalesce-
Spark includes a function to allow you to select the first non-null value from a set of columns by
using the coalesce function. 

>>>> # in Python
>>>> from pyspark.sql.functions import coalesce
>>>> df.select(coalesce(col("Description"), col("CustomerId"))).show()


ifnull, nullIf, nvl, and nvl2-
There are several other SQL functions that you can use to achieve similar things.
"ifnull" allows you to select the second value if the first is null, and defaults to the first.

"nullif" returns null if the two values are equal or else returns the second if they are not.

"nvl" returns the second value if the first is null, but defaults to the first.

"nvl2" returns the second value if the first is not null; otherwise, it will return the last specified value.

-- in SQL
SELECT
ifnull(null, 'return_value'),
nullif('value', 'value'),
nvl(null, 'return_value'),
nvl2('not_null', 'return_value', "else_value")
FROM dfTable LIMIT 1

+------------+----+------------+------------+
| a| b| c| d|
+------------+----+------------+------------+
|return_value|null|return_value|return_value|
+------------+----+------------+------------+

Naturally, we can use these in select expressions on DataFrames, as well.


drop-

"drop" removes rows that contain nulls. The default is to drop row in which any value is null:

>>>> df.na.drop()
>>>> df.na.drop("any")  	# Specifying "any" as an argument drops a row if any of the values are null

In SQL, we have to do this column by column:

-- in SQL
SELECT * FROM dfTable WHERE Description IS NOT NULL


>>>> df.na.drop("all") 		# Using “all” drops the row only if all values are null or NaN for that row

We can also apply this to certain sets of columns by passing in an array of columns:
>>>> # in Python
>>>> df.na.drop("all", subset=["StockCode", "InvoiceNo"])


fill-
Using the "fill" function, you can fill one or more columns with a set of values.

For example, to fill all null values in columns of type String, you might specify the following:
>>>> df.na.fill("All Null values become this string")

We could do the same for columns of type Integer by using df.na.fill(5:Integer), or for
Doubles df.na.fill(5:Double).To specify columns, we just pass in an array of column names like we did in the previous example:
>>>> # in Python
>>>> df.na.fill("all", subset=["StockCode", "InvoiceNo"])

>>>> # in Python
>>>> fill_cols_vals = {"StockCode": 5, "Description" : "No Value"}
>>>> df.na.fill(fill_cols_vals)


replace-
The most common use case is to replace all values in a certain column according to their current value.

>>>> # in Python
>>>> df.na.replace([""], ["UNKNOWN"], "Description")


Ordering-
As we discussed in Chapter 5, you can use "asc_nulls_first", "desc_nulls_first",
"asc_nulls_last", or "desc_nulls_last" to specify where you would like your null values to
appear in an ordered DataFrame.


Working with Complex Types-
Complex types can help you organize and structure your data in ways that make more sense for
the problem that you are hoping to solve. 
There are three kinds of complex types: structs, arrays and maps


Structs-
You can think of structs as DataFrames within DataFrames.

We can create a struct by wrapping a set of columns in parenthesis in a query:

>>>> df.selectExpr("(Description, InvoiceNo) as complex", "*")
>>>> df.selectExpr("struct(Description, InvoiceNo) as complex", "*")

>>>> # in Python
>>>> from pyspark.sql.functions import struct
>>>> complexDF = df.select(struct("Description", "InvoiceNo").alias("complex"))
>>>> complexDF.createOrReplaceTempView("complexDF")


We now have a DataFrame with a column complex. We can query it just as we might another
DataFrame, the only difference is that we use a dot syntax to do so, or the column method "getField":

>>>> complexDF.select("complex.Description")
>>>> complexDF.select(col("complex").getField("Description")

We can also query all values in the struct by using "*". This brings up all the columns to the toplevel DataFrame:

>>>> complexDF.select("complex.*")

-- in SQL
SELECT complex.* FROM complexDF


Arrays-

To define arrays, let’s work through a use case. With our current data, our objective is to take
every single word in our [Description] column and convert that into a row in our DataFrame.
The first task is to turn our [Description] column into a complex type, an array.

split-
We do this by using the split function and specify the delimiter:

>>>> # in Python
>>>> from pyspark.sql.functions import split
>>>> df.select(split(col("Description"), " ")).show(2)

-- in SQL
SELECT split(Description, ' ') FROM dfTable

+---------------------+
|split(Description, )|
+---------------------+
| [WHITE, HANGING, ...|
| [WHITE, METAL, LA...|
+---------------------+


We can also query the values of the array using Python-like syntax:

>>>> # in Python
>>>> df.select(split(col("Description"), " ").alias("array_col")).selectExpr("array_col[0]").show(2)

-- in SQL
SELECT split(Description, ' ')[0] FROM dfTable

This gives us the following result:
+------------+
|array_col[0]|
+------------+
| WHITE|
| WHITE|
+------------+


Array Length-
We can determine the array’s length by querying for its size:

>>>> # in Python
>>>> from pyspark.sql.functions import size
>>>> df.select(size(split(col("Description"), " "))).show(2) # shows 5 and 3


array_contains-
We can also see whether this array contains a value:

>>>> # in Python
>>>> from pyspark.sql.functions import array_contains
>>>> df.select(array_contains(split(col("Description"), " "), "WHITE")).show(2)

-- in SQL
SELECT array_contains(split(Description, ' '), 'WHITE') FROM dfTable

This gives us the following result:
+--------------------------------------------+
|array_contains(split(Description, ), WHITE)|
+--------------------------------------------+
| true|
| true|
+--------------------------------------------+


explode-
The "explode" function takes a column that consists of arrays and creates one row (with the rest of
the values duplicated) per value in the array.

"Hello World", "Other Col" ---->(split)---->["Hello", "World"],"Other Col" -----> (explode) ----->"Hello", "Other Col", "World", "Other Col"

>>>> # in Python
>>>> from pyspark.sql.functions import split, explode
>>>> df.withColumn("splitted", split(col("Description"), " "))\
>>>> .withColumn("exploded", explode(col("splitted")))\
>>>> .select("Description", "InvoiceNo", "exploded").show(2)

-- in SQL
SELECT Description, InvoiceNo, exploded
FROM (SELECT *, split(Description, " ") as splitted FROM dfTable)
LATERAL VIEW explode(splitted) as exploded

This gives us the following result:
+--------------------+---------+--------+
| Description|InvoiceNo|exploded|
+--------------------+---------+--------+
|WHITE HANGING HEA...| 536365| WHITE|
|WHITE HANGING HEA...| 536365| HANGING|
+--------------------+---------+--------+


To start from pg. no 115

Maps-

Maps are created by using the map function and key-value pairs of columns. You then can select
them just like you might select from an array.

>>>> # in Python
>>>> from pyspark.sql.functions import create_map
>>>> df.select(create_map(col("Description"), col("InvoiceNo")).alias("complex_map"))\
>>>> .show(2)

-- in SQL
SELECT map(Description, InvoiceNo) as complex_map FROM dfTable
WHERE Description IS NOT NULL

This produces the following result:
+--------------------+
| complex_map|
+--------------------+
|Map(WHITE HANGING...|
|Map(WHITE METAL L...|
+--------------------+
You can query them by using the proper key. A missing key returns "null".


>>>> # in Python
>>>> df.select(map(col("Description"), col("InvoiceNo")).alias("complex_map"))\
>>>> .selectExpr("complex_map['WHITE METAL LANTERN']").show(2)

This gives us the following result:
+--------------------------------+
|complex_map[WHITE METAL LANTERN]|
+--------------------------------+
| null|
| 536365|
+--------------------------------+

You can also explode map types, which will turn them into columns:

>>>> # in Python
>>>> df.select(map(col("Description"), col("InvoiceNo")).alias("complex_map"))\
>>>> .selectExpr("explode(complex_map)").show(2)

This gives us the following result:
+--------------------+------+
| key| value|
+--------------------+------+
|WHITE HANGING HEA...|536365|
| WHITE METAL LANTERN|536365|
+--------------------+------+


Working with JSON-
Spark has some unique support for working with JSON data. You can operate directly on strings
of JSON in Spark and parse from JSON or extract JSON objects. Let’s begin by creating a JSON
column:

>>>> # in Python
>>>> jsonDF = spark.range(1).selectExpr("""
>>>> '{"myJSONKey" : {"myJSONValue" : [1, 2, 3]}}' as jsonString""")


You can use the "get_json_object" to inline query a JSON object, be it a dictionary or array.
You can use "json_tuple" if this object has only one level of nesting:

>>>> # in Python
>>>> from pyspark.sql.functions import get_json_object, json_tuple
>>>> jsonDF.select(
>>>> get_json_object(col("jsonString"), "$.myJSONKey.myJSONValue[1]") as "column",
>>>> json_tuple(col("jsonString"), "myJSONKey")).show(2)

Here’s the equivalent in SQL:

jsonDF.selectExpr(
"json_tuple(jsonString, '$.myJSONKey.myJSONValue[1]') as column").show(2)

This results in the following table:
+------+--------------------+
|column| c0|
+------+--------------------+
| 2|{"myJSONValue":[1...|
+------+--------------------+

You can also turn a StructType into a JSON string by using the "to_json" function.

>>>> # in Python
>>>> from pyspark.sql.functions import to_json
>>>> df.selectExpr("(InvoiceNo, Description) as myStruct")\
>>>> .select(to_json(col("myStruct")))


This function also accepts a dictionary (map) of parameters that are the same as the JSON data
source. You can use the from_json function to parse this (or other JSON data) back in.

# in Python
from pyspark.sql.functions import from_json
from pyspark.sql.types import *
parseSchema = StructType((
StructField("InvoiceNo",StringType(),True),
StructField("Description",StringType(),True)))
df.selectExpr("(InvoiceNo, Description) as myStruct")\
.select(to_json(col("myStruct")).alias("newJSON"))\
.select(from_json(col("newJSON"), parseSchema), col("newJSON")).show(2)

This gives us the following result:
+----------------------+--------------------+
|jsontostructs(newJSON)| newJSON|
+----------------------+--------------------+
| [536365,WHITE HAN...|{"InvoiceNo":"536...|
| [536365,WHITE MET...|{"InvoiceNo":"536...|
+----------------------+--------------------+


User-Defined Functions-

These user-defined functions (UDFs) make it possible for you to write your own custom
transformations using Python or Scala and even use external libraries.
They’re just functions that operate on the data, record by record. By default, these functions 
are registered as temporary functions to be used in that specific SparkSession or Context.

Although you can write UDFs in Scala, Python, or Java, there are performance considerations
that you should be aware of. To illustrate this, we’re going to walk through exactly what happens
when you create UDF, pass that into Spark, and then execute code using that UDF.

The first step is the actual function. We’ll create a simple one for this example. Let’s write a
"power3" function that takes a number and raises it to a power of three:

>>>> # in Python
>>>> udfExampleDF = spark.range(5).toDF("num")
>>>> def power3(double_value):
>>>> return double_value ** 3
>>>> power3(2.0)

In this trivial example, we can see that our functions work as expected. We are able to provide an
individual input and produce the expected result (with this simple test case). 
Thus far, our expectations for the input are high: it must be a specific type and cannot be a null value

Now that we’ve created these functions and tested them, we need to register them with Spark so
that we can use them on all of our worker machines. Spark will serialize the function on the
driver and transfer it over the network to all executor processes. This happens regardless of
language.


If the function is written in Python, Spark starts a Python
process on the worker, serializes all of the data to a format that Python can understand
(remember, it was in the JVM earlier), executes the function row by row on that data in the
Python process, and then finally returns the results of the row operations to the JVM and Spark.

First, we need to register the function to make it available as a DataFrame function:

>>>> # in Python
>>>> from pyspark.sql.functions import udf
>>>> power3udf = udf(power3)

Then, we can use it in our DataFrame code:

>>>> # in Python
>>>> from pyspark.sql.functions import col
>>>> udfExampleDF.select(power3udf(col("num"))).show(2)

+-----------+
|power3(num)|
+-----------+
| 0|
| 1|
+-----------+

At this juncture, we can use this only as a DataFrame function. That is to say, we can’t use it
within a string expression, only on an expression. However, we can also register this UDF as a
Spark SQL function.

Because this function is registered with Spark SQL—and we’ve learned that any Spark SQL
function or expression is valid to use as an expression when working with DataFrames—we can
turn around and use the UDF in Python.

However, rather than using it as a DataFrame function, we use it as a SQL expression:

>>>> # in Python
>>>> udfExampleDF.selectExpr("power3(num)").show(2)

We can also register our Python function to be available as a SQL function and use that in any
language, as well.

As we saw in the beginning of this section, Spark manages its own type information, which
does not align exactly with Python’s types. Therefore, it’s a best practice to define the return type
for your function when you define it. It is important to note that specifying the return type is not
necessary, but it is a best practice.

If you specify the type that doesn’t align with the actual type returned by the function, Spark will
not throw an error but will just return "null" to designate a failure. You can see this if you were to
switch the return type in the following function to be a DoubleType:

>>>> # in Python
>>>> from pyspark.sql.types import IntegerType, DoubleType
>>>> spark.udf.register("power3py", power3, DoubleType())

>>>> # in Python
>>>> udfExampleDF.selectExpr("power3py(num)").show(2)
>>>> # registered via Python

This is because the range creates integers. When integers are operated on in Python, Python
won’t convert them into floats (the corresponding type to Spark’s double type), therefore we see
null. We can remedy this by ensuring that our Python function returns a float instead of an
integer and the function will behave correctly.
Naturally, we can use either of these from SQL, too, after we register them:

-- in SQL
SELECT power3(12), power3py(12) -- doesn't work because of return type

When you want to optionally return a value from a UDF, you should return "None" in Python.

_______________________NO IDEA__________________________________________________________________|
                                                                                                |
As a last note, you can also use UDF/UDAF creation via a Hive syntax. To allow for this, first	|
you must enable Hive support when they create their SparkSession (via                           |
SparkSession.builder().enableHiveSupport()). Then you can register UDFs in SQL. This		|
is only supported with precompiled Scala and Java packages, so you’ll need to specify them as a |
dependency:                                                                                     |
-- in SQL                                                                                       |
CREATE TEMPORARY FUNCTION myFunc AS 'com.organization.hive.udf.FunctionName'			|
Additionally, you can register this as a permanent function in the Hive Metastore by removing	|
TEMPORARY.											|
________________________________________________________________________________________________|



Chapter 7. Aggregations-



Aggregating is the act of collecting something together and is a cornerstone of big data analytics.
In an aggregation, you will specify a key or grouping and an aggregation function that specifies
how you should transform one or more columns.

In addition to working with any type of values, Spark also allows us to create the following
groupings types:

The simplest grouping is to just summarize a complete DataFrame by performing an
aggregation in a select statement.

A “group by” allows you to specify one or more keys as well as one or more
aggregation functions to transform the value columns.

A “window” gives you the ability to specify one or more keys as well as one or more
aggregation functions to transform the value columns. However, the rows input to the
function are somehow related to the current row.

A “grouping set,” which you can use to aggregate at multiple different levels. Grouping
sets are available as a primitive in SQL and via rollups and cubes in DataFrames.

A “rollup” makes it possible for you to specify one or more keys as well as one or more
aggregation functions to transform the value columns, which will be summarized
hierarchically.

A “cube” allows you to specify one or more keys as well as one or more aggregation
functions to transform the value columns, which will be summarized across all
combinations of columns.

Each grouping returns a "RelationalGroupedDataset" on which we specify our aggregations.

Lets read the data first

>>>> # in Python
>>>> df = spark.read.format("csv")\
>>>> .option("header", "true")\
>>>> .option("inferSchema", "true")\
>>>> .load("/data/retail-data/all/*.csv")\
>>>> .coalesce(5)
>>>> df.cache()
>>>> df.createOrReplaceTempView("dfTable")

Here’s a sample of the data so that you can reference the output of some of the functions:
+---------+---------+--------------------+--------+--------------+---------+-----
|InvoiceNo|StockCode| Description|Quantity| InvoiceDate|UnitPrice|Cu...
+---------+---------+--------------------+--------+--------------+---------+-----
| 536365| 85123A|WHITE HANGING... | 6|12/1/2010 8:26| 2.55| ...
| 536365| 71053|WHITE METAL... | 6|12/1/2010 8:26| 3.39| ...
...
| 536367| 21755|LOVE BUILDING BLO...| 3|12/1/2010 8:34| 5.95| ...
| 536367| 21777|RECIPE BOX WITH M...| 4|12/1/2010 8:34| 7.95| ...
+---------+---------+--------------------+--------+--------------+---------+-----



Aggregation Functions-

You can find most aggregation functions in the "org.apache.spark.sql.functions" package.


count-
The first function worth going over is "count", except in this example it will perform as a
transformation instead of an action. In this case, we can do one of two things: specify a specific
column to count, or all the columns by using count(*) or count(1) to represent that we want to
count every row as the literal one.

>>>> # in Python
>>>> from pyspark.sql.functions import count
>>>> df.select(count("StockCode")).show() # 541909

-- in SQL
SELECT COUNT(*) FROM dfTable


countDistinct-
If the number of unique groups that you want. To get this number, you can use the "countDistinct" function.

>>>> # in Python
>>>> from pyspark.sql.functions import countDistinct
>>>> df.select(countDistinct("StockCode")).show() # 4070

-- in SQL
SELECT COUNT(DISTINCT *) FROM DFTABLE


approx_count_distinct-
Often, we find ourselves working with large datasets and the exact distinct count is irrelevant.
There are times when an approximation to a certain degree of accuracy will work just fine, and
for that, you can use the "approx_count_distinct" function.

>>>> # in Python
>>>> from pyspark.sql.functions import approx_count_distinct
>>>> df.select(approx_count_distinct("StockCode", 0.1)).show() # 3364

-- in SQL
SELECT approx_count_distinct(StockCode, 0.1) FROM DFTABLE

You will notice that approx_count_distinct took another parameter with which you can
specify the maximum estimation error allowed. In this case, we specified a rather large error and
thus receive an answer that is quite far off but does complete more quickly than countDistinct.


first and last-
You can get the first and last values from a DataFrame by using these two obviously named
functions.

>>>> # in Python
>>>> from pyspark.sql.functions import first, last
>>>> df.select(first("StockCode"), last("StockCode")).show()

-- in SQL
SELECT first(StockCode), last(StockCode) FROM dfTable
+-----------------------+----------------------+
|first(StockCode, false)|last(StockCode, false)|
+-----------------------+----------------------+
| 85123A| 22138|
+-----------------------+----------------------+


min and max-
To extract the minimum and maximum values from a DataFrame, use the "min" and "max" functions:

>>>> # in Python
>>>> from pyspark.sql.functions import min, max
>>>> df.select(min("Quantity"), max("Quantity")).show()

-- in SQL
SELECT min(Quantity), max(Quantity) FROM dfTable

+-------------+-------------+
|min(Quantity)|max(Quantity)|
+-------------+-------------+
| -80995| 80995|
+-------------+-------------+

sum
Another simple task is to add all the values in a row using the "sum" function:

>>>> # in Python
>>>> from pyspark.sql.functions import sum
>>>> df.select(sum("Quantity")).show() # 5176450

-- in SQL
SELECT sum(Quantity) FROM dfTable


sumDistinct-
In addition to summing a total, you also can sum a distinct set of values by using the
"sumDistinct" function:

>>>> # in Python
>>>> from pyspark.sql.functions import sumDistinct
>>>> df.select(sumDistinct("Quantity")).show() # 29310

-- in SQL
SELECT SUM(Quantity) FROM dfTable -- 29310


avg-
Although you can calculate average by dividing "sum" by "count", Spark provides an easier way to
get that value via the "avg" or "mean" functions.

>>>> # in Python
>>>> from pyspark.sql.functions import sum, count, avg, expr
>>>> df.select(
>>>> count("Quantity").alias("total_transactions"),
>>>> sum("Quantity").alias("total_purchases"),
>>>> avg("Quantity").alias("avg_purchases"),
>>>> expr("mean(Quantity)").alias("mean_purchases")).selectExpr("total_purchases/total_transactions","avg_purchases","mean_purchases").show()


+--------------------------------------+----------------+----------------+
|(total_purchases / total_transactions)| avg_purchases| mean_purchases|
+--------------------------------------+----------------+----------------+
| 9.55224954743324|9.55224954743324|9.55224954743324|
+--------------------------------------+----------------+----------------+


Variance and Standard Deviation-

The variance is the average of the squared differences from the mean, and 
the standard deviation is the square root of the variance. 
These are fundamentally different statistical
formulae, and we need to differentiate between them. Spark has both the formula for the sample standard deviation as well as
the formula for the population standard deviation. By default, Spark performs the formula for
the sample standard deviation or variance if you use the "variance" or "stddev" functions.

>>>> # in Python
>>>> from pyspark.sql.functions import var_pop, stddev_pop
>>>> from pyspark.sql.functions import var_samp, stddev_samp
>>>> df.select(var_pop("Quantity"), var_samp("Quantity"),
>>>> stddev_pop("Quantity"), stddev_samp("Quantity")).show()

-- in SQL
SELECT var_pop(Quantity), var_samp(Quantity),
stddev_pop(Quantity), stddev_samp(Quantity)
FROM dfTable

+------------------+------------------+--------------------+-------------------+
| var_pop(Quantity)|var_samp(Quantity)|stddev_pop(Quantity)|stddev_samp(Quan...|
+------------------+------------------+--------------------+-------------------+
|47559.303646609056|47559.391409298754| 218.08095663447796| 218.081157850...|
+------------------+------------------+--------------------+-------------------+


skewness and kurtosis-

# in Python
from pyspark.sql.functions import skewness, kurtosis
df.select(skewness("Quantity"), kurtosis("Quantity")).show()

-- in SQL
SELECT skewness(Quantity), kurtosis(Quantity) FROM dfTable

+-------------------+------------------+
| skewness(Quantity)|kurtosis(Quantity)|
+-------------------+------------------+
|-0.2640755761052562|119768.05495536952|
+-------------------+------------------+

Covariance and Correlation-
We discussed single column aggregations, but some functions compare the interactions of the
values in two difference columns together. Two of these functions are "cov" and "corr", for
covariance and correlation.

NOTE- Correlation measures the Pearson correlation coefficient, which is scaled between –1 and +1. The covariance is scaled according to the inputs in the data.

Like the "var" function, covariance can be calculated either as the sample covariance or the
population covariance. Therefore it can be important to specify which formula you want to use.

>>>> # in Python
>>>> from pyspark.sql.functions import corr, covar_pop, covar_samp
>>>> df.select(corr("InvoiceNo", "Quantity"), covar_samp("InvoiceNo", "Quantity"),
>>>> covar_pop("InvoiceNo", "Quantity")).show()

-- in SQL
SELECT corr(InvoiceNo, Quantity), covar_samp(InvoiceNo, Quantity),
covar_pop(InvoiceNo, Quantity)
FROM dfTable
+-------------------------+-------------------------------+---------------------+
|corr(InvoiceNo, Quantity)|covar_samp(InvoiceNo, Quantity)|covar_pop(InvoiceN...|
+-------------------------+-------------------------------+---------------------+
| 4.912186085635685E-4| 1052.7280543902734| 1052.7...|
+-------------------------+-------------------------------+---------------------+


Aggregating to Complex Types-
In Spark, you can perform aggregations not just of numerical values using formulas, you can also
perform them on complex types. For example, we can collect a list of values present in a given
column or only the unique values by collecting to a set.

>>>> # in Python
>>>> from pyspark.sql.functions import collect_set, collect_list
>>>> df.agg(collect_set("Country"), collect_list("Country")).show()

-- in SQL
SELECT collect_set(Country), collect_set(Country) FROM dfTable

+--------------------+---------------------+
|collect_set(Country)|collect_list(Country)|
+--------------------+---------------------+
|[Portugal, Italy,...| [United Kingdom, ...|
+--------------------+---------------------+



Grouping-

A more common task is to perform calculations based on groups in the data. 
This is typically done on categorical data for
which we group our data on one column and perform some calculations on the other columns
that end up in that group.

The first will be a count, just as we did before

df.groupBy("InvoiceNo", "CustomerId").count().show()


-- in SQL
SELECT count(*) FROM dfTable GROUP BY InvoiceNo, CustomerId

+---------+----------+-----+
|InvoiceNo|CustomerId|count|
+---------+----------+-----+
| 536846| 14573| 76|
...
| C544318| 12989| 1|
+---------+----------+-----+


Grouping with Expressions-

Rather than passing that function as an expression
into a "select" statement, we specify it as within "agg". This makes it possible for you to pass-in
arbitrary expressions that just need to have some aggregation specified.

>>>> # in Python
>>>> from pyspark.sql.functions import count
>>>> df.groupBy("InvoiceNo").agg(
>>>> count("Quantity").alias("quan"),
>>>> expr("count(Quantity)")).show()

+---------+----+---------------+
|InvoiceNo|quan|count(Quantity)|
+---------+----+---------------+
| 536596| 6| 6|
...
| C542604| 8| 8|
+---------+----+---------------+


Grouping with Maps-
Sometimes, it can be easier to specify your transformations as a series of Maps for which the key
is the column, and the value is the aggregation function (as a string) that you would like to
perform. You can reuse multiple column names if you specify them inline, as well:

>>>> # in Python
>>>> df.groupBy("InvoiceNo").agg(expr("avg(Quantity)"),expr("stddev_pop(Quantity)"))\
>>>> .show()

-- in SQL
SELECT avg(Quantity), stddev_pop(Quantity), InvoiceNo FROM dfTable
GROUP BY InvoiceNo

+---------+------------------+--------------------+
|InvoiceNo| avg(Quantity)|stddev_pop(Quantity)|
+---------+------------------+--------------------+
| 536596| 1.5| 1.1180339887498947|
...
| C542604| -8.0| 15.173990905493518|
+---------+------------------+--------------------




Window Functions-

Spark supports three kinds of window functions: ranking functions, analytic functions,
and aggregate functions.

To demonstrate, we will add a date column that will convert our invoice date into a column that
contains only date information (not time information, too):

>>>> # in Python
>>>> from pyspark.sql.functions import col, to_date
>>>> dfWithDate = df.withColumn("date", to_date(col("InvoiceDate"), "MM/d/yyyy H:mm"))
>>>> dfWithDate.createOrReplaceTempView("dfWithDate")

The first step to a window function is to create a window specification. Note that the "partition
by" is unrelated to the partitioning scheme concept that we have covered thus far. It’s just a
similar concept that describes how we will be breaking up our group. The ordering determines
the ordering within a given partition, and, finally, the frame specification (the "rowsBetween"
statement) states which rows will be included in the frame based on its reference to the current
input row. In the following example, we look at all previous rows up to the current row:


>>>> # in Python
>>>> from pyspark.sql.window import Window
>>>> from pyspark.sql.functions import desc
>>>> windowSpec = Window\
>>>> .partitionBy("CustomerId", "date")\
>>>> .orderBy(desc("Quantity"))\
>>>> .rowsBetween(Window.unboundedPreceding, Window.currentRow)


Now we want to use an aggregation function to learn more about each specific customer. 
For now, establishing the maximum purchase quantity over all time.

>>>> # in Python
>>>> from pyspark.sql.functions import max
>>>> maxPurchaseQuantity = max(col("Quantity")).over(windowSpec)


You will notice that this returns a column (or expressions). We can now use this in a DataFrame
select statement

>>>> # in Python
>>>> from pyspark.sql.functions import dense_rank, rank
>>>> purchaseDenseRank = dense_rank().over(windowSpec)
>>>> purchaseRank = rank().over(windowSpec)

This also returns a column that we can use in select statements. Now we can perform a select to
view the calculated window values:

>>>> # in Python
>>>> from pyspark.sql.functions import col
>>>> dfWithDate.where("CustomerId IS NOT NULL").orderBy("CustomerId")\
>>>> .select(
>>>> col("CustomerId"),
>>>> col("date"),
>>>> col("Quantity"),
>>>> purchaseRank.alias("quantityRank"),
>>>> purchaseDenseRank.alias("quantityDenseRank"),
>>>> maxPurchaseQuantity.alias("maxPurchaseQuantity")).show()

-- in SQL
SELECT CustomerId, date, Quantity,
rank(Quantity) OVER (PARTITION BY CustomerId, date
ORDER BY Quantity DESC NULLS LAST
ROWS BETWEEN
UNBOUNDED PRECEDING AND
CURRENT ROW) as rank,
dense_rank(Quantity) OVER (PARTITION BY CustomerId, date
ORDER BY Quantity DESC NULLS LAST
ROWS BETWEEN
UNBOUNDED PRECEDING AND
CURRENT ROW) as dRank,
max(Quantity) OVER (PARTITION BY CustomerId, date
ORDER BY Quantity DESC NULLS LAST
ROWS BETWEEN
UNBOUNDED PRECEDING AND
CURRENT ROW) as maxPurchase
FROM dfWithDate WHERE CustomerId IS NOT NULL ORDER BY CustomerId

+----------+----------+--------+------------+-----------------+---------------+
|CustomerId| date|Quantity|quantityRank|quantityDenseRank|maxP...Quantity|
+----------+----------+--------+------------+-----------------+---------------+
| 12346|2011-01-18| 74215| 1| 1| 74215|
| 12346|2011-01-18| -74215| 2| 2| 74215|
| 12347|2010-12-07| 36| 1| 1| 36|
| 12347|2010-12-07| 30| 2| 2| 36|
...
| 12347|2010-12-07| 12| 4| 4| 36|
| 12347|2010-12-07| 6| 17| 5| 36|
| 12347|2010-12-07| 6| 17| 5| 36|
+----------+----------+--------+------------+-----------------+---------------



Grouping Sets-

Sometimes we want something a bit
more complete—an aggregation across multiple groups. We achieve this by using grouping sets.
Grouping sets are a low-level tool for combining sets of aggregations together. They give you the
ability to create arbitrary aggregation in their group-by statements.

In below eg., we would like to get the
total quantity of all stock codes and customers. To do so, we’ll use the following SQL
expression:

>>>> # in Python
>>>> dfNoNull = dfWithDate.drop()
>>>> dfNoNull.createOrReplaceTempView("dfNoNull")

-- in SQL
SELECT CustomerId, stockCode, sum(Quantity) FROM dfNoNull
GROUP BY customerId, stockCode
ORDER BY CustomerId DESC, stockCode DESC

+----------+---------+-------------+
|CustomerId|stockCode|sum(Quantity)|
+----------+---------+-------------+
| 18287| 85173| 48|
| 18287| 85040A| 48|
| 18287| 85039B| 120|
...
| 18287| 23269| 36|
+----------+---------+-------------+

You can do the exact same thing by using a grouping set:

-- in SQL
SELECT CustomerId, stockCode, sum(Quantity) FROM dfNoNull
GROUP BY customerId, stockCode GROUPING SETS((customerId, stockCode))
ORDER BY CustomerId DESC, stockCode DESC

+----------+---------+-------------+
|CustomerId|stockCode|sum(Quantity)|
+----------+---------+-------------+
| 18287| 85173| 48|
| 18287| 85040A| 48|
| 18287| 85039B| 120|
...
| 18287| 23269| 36|
+----------+---------+-------------+

Simple enough, but what if you also want to include the total number of items, regardless of
customer or stock code? With a conventional group-by statement, this would be impossible. But,
it’s simple with grouping sets: we simply specify that we would like to aggregate at that level, as
well, in our grouping set. This is, effectively, the union of several different groupings together:

-- in SQL
SELECT CustomerId, stockCode, sum(Quantity) FROM dfNoNull
GROUP BY customerId, stockCode GROUPING SETS((customerId, stockCode),())
ORDER BY CustomerId DESC, stockCode DESC

+----------+---------+-------------+
|customerId|stockCode|sum(Quantity)|
+----------+---------+-------------+
| 18287| 85173| 48|
| 18287| 85040A| 48|
| 18287| 85039B| 120|
...
| 18287| 23269| 36|
+----------+---------+-------------+


The GROUPING SETS operator is only available in SQL. To perform the same in DataFrames, you
use the "rollup" and "cube" operators—which allow us to get the same results.


Rollups-
A rollup is a multidimensional aggregation that performs a variety of group-by style calculations
for us.

Let’s create a rollup-

>>>> # in Python
>>>> rolledUpDF = dfNoNull.rollup("Date", "Country").agg(sum("Quantity"))\
>>>> .selectExpr("Date", "Country", "`sum(Quantity)` as total_quantity")\
>>>> .orderBy("Date")
>>>> rolledUpDF.show()

+----------+--------------+--------------+
| Date| Country|total_quantity|
+----------+--------------+--------------+
| null| null| 5176450|
|2010-12-01|United Kingdom| 23949|
|2010-12-01| Germany| 117|
|2010-12-01| France| 449|
...
|2010-12-03| France| 239|
|2010-12-03| Italy| 164|
|2010-12-03| Belgium| 528|
+----------+--------------+--------------+

Now where you see the null values is where you’ll find the grand totals. A null in both rollup
columns specifies the grand total across both of those columns:

>>>> rolledUpDF.where("Country IS NULL").show()
>>>> rolledUpDF.where("Date IS NULL").show()

+----+-------+--------------+
|Date|Country|total_quantity|
+----+-------+--------------+
|null| null| 5176450|
+----+-------+--------------+



Cube-
A "cube" takes the "rollup" to a level deeper. Rather than treating elements hierarchically, a "cube"
does the same thing across all dimensions.

The method call is quite similar, but instead of calling rollup, we call cube:

>>>> # in Python
>>>> from pyspark.sql.functions import sum
>>>> dfNoNull.cube("Date", "Country").agg(sum(col("Quantity")))\
>>>> .select("Date", "Country", "sum(Quantity)").orderBy("Date").show()

+----+--------------------+-------------+
|Date| Country|sum(Quantity)|
+----+--------------------+-------------+
|null| Japan| 25218|
|null| Portugal| 16180|
|null| Unspecified| 3300|
|null| null| 5176450|
|null| Australia| 83653|
...
|null| Norway| 19247|
|null| Hong Kong| 4769|
|null| Spain| 26824|
|null| Czech Republic| 592|
+----+--------------------+-------------+

This is a quick and easily accessible summary of nearly all of the information in our table, and
it’s a great way to create a quick summary table that others can use later on.


Pivot-
Pivots make it possible for you to convert a row into a column. 

>>>> # in Python
>>>> pivoted = dfWithDate.groupBy("date").pivot("Country").sum()

This DataFrame will now have a column for every combination of country, numeric variable,
and a column specifying the date.

For example, for USA we have the following columns:
USA_sum(Quantity), USA_sum(UnitPrice), USA_sum(CustomerID). This represents one for
each numeric column in our dataset (because we just performed an aggregation over all of them).
Here’s an example query and result from this data.

>>>> pivoted.where("date > '2011-12-05'").select("date" ,"`USA_sum(Quantity)`").show()

+----------+-----------------+
| date|USA_sum(Quantity)|
+----------+-----------------+
|2011-12-06| null|
|2011-12-09| null|
|2011-12-08| -196|
|2011-12-07| null|
+----------+-----------------+






CHAPTER 7 - Joins-


Join Expressions-

A join brings together two sets of data, the left and the right, by comparing the value of one or
more keys of the left and right and evaluating the result of a join expression that determines
whether Spark should bring together the left set of data with the right set of data. The most
common join expression, an equi-join, compares whether the specified keys in your left and
right datasets are equal.

Join Types-

Inner joins (keep rows with keys that exist in the left and right datasets)

Outer joins (keep rows with keys in either the left or right datasets)

Left outer joins (keep rows with keys in the left dataset)

Right outer joins (keep rows with keys in the right dataset)

Left semi joins (keep the rows in the left, and only the left, dataset where the key appears in the right dataset)

Left anti joins (keep the rows in the left, and only the left, dataset where they do not appear in the right dataset)

Natural joins (perform a join by implicitly matching the columns between the two datasets with the same names)

Cross (or Cartesian) joins (match every row in the left dataset with every row in the right dataset)

let’s create some simple datasets that we can use in our examples:

>>>> # in Python
>>>> person = spark.createDataFrame([
>>>> (0, "Bill Chambers", 0, [100]),
>>>> (1, "Matei Zaharia", 1, [500, 250, 100]),
>>>> (2, "Michael Armbrust", 1, [250, 100])])\
>>>> .toDF("id", "name", "graduate_program", "spark_status")

>>>> graduateProgram = spark.createDataFrame([
>>>> (0, "Masters", "School of Information", "UC Berkeley"),
>>>> (2, "Masters", "EECS", "UC Berkeley"),
>>>> (1, "Ph.D.", "EECS", "UC Berkeley")])\
>>>> .toDF("id", "degree", "department", "school")

>>>> sparkStatus = spark.createDataFrame([
>>>> (500, "Vice President"),
>>>> (250, "PMC Member"),
>>>> (100, "Contributor")])\
>>>> .toDF("id", "status")


Next, let’s register these as tables so that we use them throughout the chapter:

>>>> person.createOrReplaceTempView("person")
>>>> graduateProgram.createOrReplaceTempView("graduateProgram")
>>>> sparkStatus.createOrReplaceTempView("sparkStatus")


Inner Joins-

Inner joins evaluate the keys in both of the DataFrames or tables and include (and join together)
only the rows that evaluate to true. Here, we join the graduateProgram
DataFrame with the person DataFrame to create a new DataFrame:

>>>> ## in Python
>>>> joinExpression = person["graduate_program"] == graduateProgram['id']

Keys that do not exist in both DataFrames will not show in the resulting DataFrame. For
example, the following expression would result in zero values in the resulting DataFrame:

>>>> wrongJoinExpression = person["name"] == graduateProgram["school"]

Inner joins are the default join, so we just need to specify our left DataFrame and join the right in
the JOIN expression:

>>>> person.join(graduateProgram, joinExpression).show()

-- in SQL
SELECT * FROM person JOIN graduateProgram
ON person.graduate_program = graduateProgram.id

+---+----------------+----------------+---------------+---+-------+----------+---
| id| name|graduate_program| spark_status| id| degree|department|...
+---+----------------+----------------+---------------+---+-------+----------+---
| 0| Bill Chambers| 0| [100]| 0|Masters| School...|...
| 1| Matei Zaharia| 1|[500, 250, 100]| 1| Ph.D.| EECS|...
| 2|Michael Armbrust| 1| [250, 100]| 1| Ph.D.| EECS|...
+---+----------------+----------------+---------------+---+-------+----------+---

We can also specify this explicitly by passing in a third parameter, the joinType:

>>>> # in Python
>>>> joinType = "inner"
>>>> person.join(graduateProgram, joinExpression, joinType).show()

-- in SQL
SELECT * FROM person INNER JOIN graduateProgram
ON person.graduate_program = graduateProgram.id

+---+----------------+----------------+---------------+---+-------+--------------
| id| name|graduate_program| spark_status| id| degree| department...
+---+----------------+----------------+---------------+---+-------+--------------
| 0| Bill Chambers| 0| [100]| 0|Masters| School...
| 1| Matei Zaharia| 1|[500, 250, 100]| 1| Ph.D.| EECS...
| 2|Michael Armbrust| 1| [250, 100]| 1| Ph.D.| EECS...
+---+----------------+----------------+---------------+---+-------+--------------


Outer Joins-

Outer joins evaluate the keys in both of the DataFrames or tables and includes (and joins
together) the rows that evaluate to true or false. If there is no equivalent row in either the left or
right DataFrame, Spark will insert null:

>>>> joinType = "outer"
>>>> person.join(graduateProgram, joinExpression, joinType).show()

-- in SQL
SELECT * FROM person FULL OUTER JOIN graduateProgram
ON graduate_program = graduateProgram.id

+----+----------------+----------------+---------------+---+-------+-------------
| id| name|graduate_program| spark_status| id| degree| departmen...
+----+----------------+----------------+---------------+---+-------+-------------
| 1| Matei Zaharia| 1|[500, 250, 100]| 1| Ph.D.| EEC...
| 2|Michael Armbrust| 1| [250, 100]| 1| Ph.D.| EEC...
|null| null| null| null| 2|Masters| EEC...
| 0| Bill Chambers| 0| [100]| 0|Masters| School...
+----+----------------+----------------+---------------+---+-------+-------------


Left Outer Joins-

Left outer joins evaluate the keys in both of the DataFrames or tables and includes all rows from
the left DataFrame as well as any rows in the right DataFrame that have a match in the left
DataFrame. If there is no equivalent row in the RIGHT DataFrame, Spark will insert null:

>>>> joinType = "left_outer"
>>>> graduateProgram.join(person, joinExpression, joinType).show()

-- in SQL
SELECT * FROM graduateProgram LEFT OUTER JOIN person
ON person.graduate_program = graduateProgram.id

+---+-------+----------+-----------+----+----------------+----------------+---
| id| degree|department| school| id| name|graduate_program|...
+---+-------+----------+-----------+----+----------------+----------------+---
| 0|Masters| School...|UC Berkeley| 0| Bill Chambers| 0|...
| 2|Masters| EECS|UC Berkeley|null| null| null|...
| 1| Ph.D.| EECS|UC Berkeley| 2|Michael Armbrust| 1|...
| 1| Ph.D.| EECS|UC Berkeley| 1| Matei Zaharia| 1|...
+---+-------+----------+-----------+----+----------------+----------------+---


Right Outer Joins-

Right outer joins evaluate the keys in both of the DataFrames or tables and includes all rows
from the right DataFrame as well as any rows in the left DataFrame that have a match in the right
DataFrame. If there is no equivalent row in the left DataFrame, Spark will insert null:

>>>> joinType = "right_outer"
>>>> person.join(graduateProgram, joinExpression, joinType).show()

-- in SQL
SELECT * FROM person RIGHT OUTER JOIN graduateProgram
ON person.graduate_program = graduateProgram.id

+----+----------------+----------------+---------------+---+-------+------------+
| id| name|graduate_program| spark_status| id| degree| department|
+----+----------------+----------------+---------------+---+-------+------------+
| 0| Bill Chambers| 0| [100]| 0|Masters|School of...|
|null| null| null| null| 2|Masters| EECS|
| 2|Michael Armbrust| 1| [250, 100]| 1| Ph.D.| EECS|
| 1| Matei Zaharia| 1|[500, 250, 100]| 1| Ph.D.| EECS|
+----+----------------+----------------+---------------+---+-------+------------+


Left Semi Joins-

They do not actually include any values
from the right DataFrame. They only compare values to see if the value exists in the second
DataFrame. If the value does exist, those rows will be kept in the result, even if there are
duplicate keys in the left DataFrame.

joinType = "left_semi"
graduateProgram.join(person, joinExpression, joinType).show()

+---+-------+--------------------+-----------+
| id| degree| department| school|
+---+-------+--------------------+-----------+
| 0|Masters|School of Informa...|UC Berkeley|
| 1| Ph.D.| EECS|UC Berkeley|
+---+-------+--------------------+-----------+

>>>> gradProgram2 = graduateProgram.union(spark.createDataFrame([
>>>> (0, "Masters", "Duplicated Row", "Duplicated School")]))
>>>> gradProgram2.createOrReplaceTempView("gradProgram2")
>>>> gradProgram2.join(person, joinExpression, joinType).show()

-- in SQL
SELECT * FROM gradProgram2 LEFT SEMI JOIN person
ON gradProgram2.id = person.graduate_program

+---+-------+--------------------+-----------------+
| id| degree| department| school|
+---+-------+--------------------+-----------------+
| 0|Masters|School of Informa...| UC Berkeley|
| 1| Ph.D.| EECS| UC Berkeley|
| 0|Masters| Duplicated Row|Duplicated School|
+---+-------+--------------------+-----------------+


Left Anti Joins-

Like left semi joins, they do not actually
include any values from the right DataFrame. They only compare values to see if the value exists
in the second DataFrame. However, rather than keeping the values that exist in the second
DataFrame, they keep only the values that do not have a corresponding key in the second
DataFrame

>>>> joinType = "left_anti"
>>>> graduateProgram.join(person, joinExpression, joinType).show()

-- in SQL
SELECT * FROM graduateProgram LEFT ANTI JOIN person
ON graduateProgram.id = person.graduate_program

+---+-------+----------+-----------+
| id| degree|department| school|
+---+-------+----------+-----------+
| 2|Masters| EECS|UC Berkeley|
+---+-------+----------+-----------+


Natural Joins-

Natural joins make implicit guesses at the columns on which you would like to join. It finds
matching columns and returns the results. Left, right, and outer natural joins are all supported.

+---+-------+----------+-----------+---+----------------+----------------+-------
| id| degree|department| school| id| name|graduate_program|spar...
+---+-------+----------+-----------+---+----------------+----------------+-------
| 0|Masters| School...|UC Berkeley| 0| Bill Chambers| 0| ...
| 1| Ph.D.| EECS|UC Berkeley| 2|Michael Armbrust| 1| [2...
| 1| Ph.D.| EECS|UC Berkeley| 1| Matei Zaharia| 1|[500...
+---+-------+----------+-----------+---+----------------+----------------+-------



Cross (Cartesian Joins)-

Cross-joins in simplest terms are inner
joins that do not specify a predicate. Cross joins will join every single row in the left DataFrame
to ever single row in the right DataFrame. This will cause an absolute explosion in the number of
rows contained in the resulting DataFrame. If you have 1,000 rows in each DataFrame, the crossjoin of these will result in 1,000,000 (1,000 x 1,000) rows.

>>>> joinType = "cross"
>>>> graduateProgram.join(person, joinExpression, joinType).show()

-- in SQL
SELECT * FROM graduateProgram CROSS JOIN person
ON graduateProgram.id = person.graduate_program

If you truly intend to have a cross-join, you can call that out explicitly:

>>>> person.crossJoin(graduateProgram).show()

-- in SQL
SELECT * FROM graduateProgram CROSS JOIN person

+---+----------------+----------------+---------------+---+-------+-------------+
| id| name|graduate_program| spark_status| id| degree| departm...|
+---+----------------+----------------+---------------+---+-------+-------------+
| 0| Bill Chambers| 0| [100]| 0|Masters| School...|
...
| 1| Matei Zaharia| 1|[500, 250, 100]| 0|Masters| School...|
...
| 2|Michael Armbrust| 1| [250, 100]| 0|Masters| School...|
...
+---+----------------+----------------+---------------+---+-------+-------------+


Challenges When Using Joins-

1. Joins on Complex Types-
Even though this might seem like a challenge, it’s actually not. Any expression is a valid join
expression, assuming that it returns a Boolean

>>>> # in Python
>>>> from pyspark.sql.functions import expr
>>>> person.withColumnRenamed("id", "personId")\
>>>> .join(sparkStatus, expr("array_contains(spark_status, id)")).show()

-- in SQL

SELECT * FROM
(select id as personId, name, graduate_program, spark_status FROM person)
INNER JOIN sparkStatus ON array_contains(spark_status, id)

+--------+----------------+----------------+---------------+---+--------------+
|personId| name|graduate_program| spark_status| id| status|
+--------+----------------+----------------+---------------+---+--------------+
| 0| Bill Chambers| 0| [100]|100| Contributor|
| 1| Matei Zaharia| 1|[500, 250, 100]|500|Vice President|
| 1| Matei Zaharia| 1|[500, 250, 100]|250| PMC Member|
| 1| Matei Zaharia| 1|[500, 250, 100]|100| Contributor|
| 2|Michael Armbrust| 1| [250, 100]|250| PMC Member|
| 2|Michael Armbrust| 1| [250, 100]|100| Contributor|
+--------+----------------+----------------+---------------+---+--------------+



Handling Duplicate Column Names-

One of the tricky things that come up in joins is dealing with duplicate column names in your
results DataFrame. In a DataFrame, each column has a unique ID within Spark’s SQL Engine,
Catalyst. This unique ID is purely internal and not something that you can directly reference.
This makes it quite difficult to refer to a specific column when you have a DataFrame with
duplicate column names.

This can occur in two distinct situations:
1. The join expression that you specify does not remove one key from one of the input DataFrames and the keys have the same column name

2. Two columns on which you are not performing the join have the same name.

Approach 1: Different join expression

When you have two keys that have the same name, probably the easiest fix is to change the join
expression from a Boolean expression to a string or sequence. This automatically removes one of
the columns for you during the join.

Approach 2: Dropping the column after the join
Another approach is to drop the offending column after the join. When doing this, we need to
refer to the column via the original source DataFrame. We can do this if the join uses the same
key names or if the source DataFrames have columns that simply have the same name.

Approach 3: Renaming a column before the join
We can avoid this issue altogether if we rename one of our columns before the join.