Essential PySpark: Data Engineering

Question 1

The performance gain in Distributed Computing lies in splitting the data into smaller chunks and _ processing them on _ machine(s).
(iteratively/parallely)
(single machine/several machines)

Answer 1

1. parallely
2. several

Code (containing business logic) for processing is brought to the place where the data chunk is actually stored. This technique is called “Data Parallel Processing”

Question 2

The MapReduce paradigm breaks down a Data Parallel Processing problem into three kinds of stages.
Name them.

Answer 2

1. Map stage
2. Shuffle stage
3. Reduce stage

There can be multiple Map stages followed by multiple Reduce stages. However, a Reduce
stage only starts after all of the Map stages have been completed.

Question 3

The Map stage takes the input dataset, splits it into _ .

Answer 3

(key, value) pairs

Question 4

The Shuffle stage takes the (key, value) pairs from the Map stage and _ them so that pairs with the same key end up together.

Answer 4

sorts/groups

Sorting/Grouping is based on keys.
A processing core in the cluster cannot have two non-trivial groups (with freq > 1)

Question 5

The Reduce stage takes the resultant groups of (key, value) pairs from the Shuffle stage and _ them (groups) to produce the final result.

Answer 5

reduces or aggregates

Question 6

Commodity Hardware used to store data chunks is prone to failures.
MapReduce achieves resiliency to hardware failures by saving the results of every stage to _ , which offers
* replication
* checkpointing
* re-execution

as resiliency mechanisms.

Answer 6

HDFS
(Hadoop Distributed File System)

Question 7

How does “replication of blocks” in HDFS help make mapreduce resilient?

Answer 7

Input and intermediate data chunks are stored in HDFS, which replicates blocks (default 3 copies) across different machines. If one disk fails, another copy exists elsewhere.

Question 8

How does “Checkpointing intermediate results” in HDFS help make mapreduce resilient?

Answer 8

Each stage’s output is written to disk so that if a node fails mid‑computation, the job can restart from the last completed stage instead of reprocessing everything.

Question 9

How does “Task re‑execution” in HDFS help make mapreduce resilient?

Answer 9

If a mapper/reducer fails, the mapreduce framework reassigns the task to another node having a replica of the data

Question 10

The round-trip to HDFS (disk) after every stage makes MapReduce relatively _ at
processing data because of the _ I/O performance of physical disks in general.
(slow/fast)

Answer 10

1. slow
2. slow

Question 11

To overcome the limitation of slow performance of mapreduce, Spark keeps intermediate results in memory ( _ than I/O), but still falling back to disk when needed for resiliency.
Additionally, Spark offers much more _ APIs to express data transformations as compared to mapreduce (Hadoop).

(slower/faster)
(low-level/high-level)

Answer 11

1. faster
2. high-level

Question 12

A _ is a group of computers (machines) all working together as a single unit to solve a distributed computing problem.

Answer 12

cluster

Question 13

The primary machine of a Cluster, called the _ , takes care of the orchestration and management of the Cluster.

Answer 13

Master Node

Question 14

The secondary machines in a cluster that actually perform the task (data processing code)
are called _ .

Answer 14

Worker Nodes

At any time, a Cluster has exactly one Master node and one or more Worker nodes.

Question 15

In Spark, Resilient Distributed Dataset (RDD) is _ data structure that is _ across the cluster, residing in memory (RAM) of _ .

(mutable/immutable)
(localized/distributed)
(single machine / several machines)

Answer 15

1. immutable
2. distributed
3. several machines

Question 16

An RDD consists of _ , which are logical divisions of an RDD, with a few of them residing on each machine/node of the cluster.

Answer 16

partitions

Question 17

_ are used to manipulate the data stored within the partitions of an RDD.

Hint: They operate on RDDs.

Answer 17

Higher-order Functions

map, flatMap, reduce, fold, filter, reduceByKey, join, and union to name a few.

Question 18

Higher-order function accepts _ as
parameter, which is an inner function that helps us define the actual business logic that transforms data and is applied to each partition of the RDD in parallel.

Answer 18

lambda

Note: lambdas apply (transformations) to all partitions of the RDD at once (in parallel)

Question 19

Given SparkContext sc, write PySpark program to read a text file as an RDD and split each word based on a delimiter such as a whitespace. Transform each word into tuple with default count = 1.

Answer 19

linesRDD = sc.textFile("/path/to/file")
wordsRDD = linesRDD.flatMap(lambda s: s.split())
tuplesRDD = wordsRDD.map(lambda s: (s, 1))

Question 20

At any given point in time, an RDD has information of all the individual operations performed on it, going back all the way to the data source itself.
How is this “lineage information” useful for fault tolerance?

Answer 20

If any Executors are lost due to any failures and one or more of its partitions are lost, RDD can easily recreate the lost partitions from the source data by making use of the lineage information, thus making it Resilient to failures.

Question 21

List 3 major components of Apache Spark architecture.

Answer 21

1. Driver (JVM process)
2. Executors (JVM processes)
3. Cluster Manager

Question 22

Each Executor process is launched on _ node of the Spark cluster.

(master/worker)

Answer 22

worker

Question 23

1. In _ deploy mode, the Driver runs on a worker node chosen by the cluster manager.
2. In _ deploy mode, the Driver runs on the machine where you submitted the job

In both modes, Executors still run on worker nodes of the cluster (close to distributed data)

Answer 23

1. cluster
2. client

The deploy mode is specified via flag when we submit a Spark job through spark-submit command.

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  my_app.py (or my_scala_app.jar)

Question 24

Spark supports running with a variety of clusters depending upon your scale, infrastructure, and operational needs.
The concept of a master node (for Driver) is externalized when Spark runs on
1. YARN
2. Kubernetes.
as clusters.

Name the cluster manager for each of the following clusters:
1. Standalone
2. YARN
3. Kubernetes

Answer 24

1. Spark’s built-in manager
2. YARN ResourceManager
3. K8s control plane

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  my_app.py (or my_scala_app.jar)

Question 25

With `--deploy-mode cluster` and `--master spark://`, the Driver process runs in _ .

Answer 25

a worker node

Question 26

With `--deploy-mode cluster` and `--master yarn`, the Driver process runs in _ .

Answer 26

inside YARN ApplicationMaster

Question 27

With `--deploy-mode cluster` and `--master k8s://https://`, the Driver process runs in _ .

Answer 27

in Kubernetes cluster, as a pod.

Question 28

The _ is the single point of failure for a Spark cluster, and the entire Spark application fails if the driver fails. Therefore, different Cluster Managers implement different strategies to make it highly available.

Answer 28

Driver (JVM process)

Question 29

_ is responsible for providing resources requested by the Driver. It also monitors the Executors regarding task progress and their status. ## Footnote Resources like: compute, memory, network bandwidth, storage (disc)

Answer 29

Cluster Manager

Question 30

List major sequence of events on submitting a Spark job.

Answer 30

1. You submit job (`spark-submit`) 2. Cluster Manager starts the Driver JVM. 3. Driver requests for Executors (along with their respective resources). 4. Cluster Manager launches Executors on worker nodes. 5. Driver sends tasks to Executors.

Question 31

To make Executors run your code, Spark Driver must send your function (the lambda) and any data it needs across the network. These functions/lambdas need to be converted into "byte stream" through the process called _ . The Python library used for this process is _ .

Answer 31

1. serialization 2. cloudpickle ## Footnote Executors then deserialize that byte stream back into a Python function and apply it to their local RDD partition.

Question 32

Which kind of functions are serializable?

Answer 32

Pure functions using only built-in Python types and operations.

Question 33

1. If your lambda uses only basic Python data types, math, string ops, or simple functions, it’s _ . 2. If it touches external resources, system state, or complex objects, it’s _ .

Answer 33

1. serializable 2. not serializable

Question 34

If your Spark functions/lambdas reference non-serializable state (files, sockets, threads), Spark will throw a _ or runtime error

Answer 34

PicklingError

Question 35

Name the schedulers in Spark Driver that create tasks.

Answer 35

1. TaskScheduler 2. DAGScheduler

Question 36

The Spark SQL engine supports two types of APIs, namely, _ .

Answer 36

1. DataFrame 2. Spark SQL ## Footnote the Spark SQL engine was added as a layer on top of the RDD API, offering an even easier to use and familiar API for developers.

Question 37

At API level, the difference between a Spark DataFrame and Pandas DataFrame is that a _ resides in the memory of several machines whereas a _ resides in memory of a single machine.

Answer 37

1. Spark DataFrame 2. Pandas DataFrame

Question 38

To allow you to manipulate data, DataFrames come with operations that can be grouped into two main categories, namely, _ .

Answer 38

1. transformations 2. actions

Question 39

SQL engine comes with a built-in query optimizer called _ . This optimizer analyzes user code, along with any available statistics on the data, to generate the best possible execution plan for the query, called _ . This execution plan is further converted into Java bytecode, which runs natively inside the Executor (JVM).

Answer 39

1. Catalyst 2. Query Plan

Question 40

_ are operations performed on DataFrames that manipulate the data in the DataFrame and result in another DataFrame.

Answer 40

Transformations ## Footnote Some examples of transformations are read, select, where, filter, join, and groupBy.

Question 41

_ are operations that actually cause a result to be calculated and either printed onto the console or, more practically, written back to a storage location.

Answer 41

Actions ## Footnote Some examples of actions include write, count, show and collect.

Question 42

Spark transformations are not evaluated immediately as they are declared, and data is not manifested in memory until an _ is called.

Answer 42

action ## Footnote This principle is called lazy evaluation. Spark job is triggered in response to an action.

Question 43

When any plain text file gets uploaded, its schema is read as: column: _ rows: each line as string

Answer 43

`value` ## Footnote default column name is `value`

Question 44

The Executors apply functions in `pyspark.sql.functions` to operate 1. _ within a partition 2. _ across partitions of RDD

Answer 44

1. sequentially (row by row) 2. parallely

Question 45

If RDD has 10 partitions, then _ Executors (or cores) can process them simultaneously (parallely)

Answer 45

10

Question 46

Write PySpark code to break up the physical data stored in Amazon S3 into eight partitions.

Answer 46

``` log_df = spark.read.text("/path/to/s3/bucket").repartition(8) print(log_df.rdd.getNumPartitions()) ```

Question 47

Write code to read csv file while retaining csv columns as dataframe columns.

Answer 47

``` df = spark.read \ .option('header','true') \ .csv("/path/to/file.csv") ```

Question 48

Rename column(s) of the table read by `spark.range()` ## Footnote Simple one-column rename right after creation

Answer 48

``` val df = spark.range(1, 10001, 1, 8).toDF("number") df.show(5) ``` ## Footnote Here, `toDF("number")` renames the default column `id` to `number`.

Question 49

Rename columns `id`, `name` of the table with `CustomerId`, `Cname`. ## Footnote Renaming multiple existing columns in a pipeline

Answer 49

``` df = spark.read.table("/path/to/table") .withColumnRenamed("id", "CustomerId") .withColumnRenamed("name", "Cname") ```

Question 50

Display the schema of `df` in tree format.

Answer 50

``` df.printSchema() ```

Question 51

Create dataframe with rows: `(14, "Tom"), (23, "Alice"), (16, "Bob")`

Answer 51

``` df = spark.createDataFrame([(14, "Tom"), (23, "Alice"), (16, "Bob")], ["age", "name"]) ``` ## Footnote last tuple denotes column names

Question 52

Write code to read csv file while retaining csv columns as dataframe columns and inferring datatypes from values in csv file.

Answer 52

Option 1 ``` df = spark.read \ .option('header','true') \ .csv('path/to/file', inferSchema=True) ``` Option 2 ``` df = spark.read.csv("path/to/file", header=True, inferSchema=True) ``` ## Footnote without inferSchema=True, numbers in .csv file will also be read as string.

Question 53

Return first 5 rows of dataframe `df`

Answer 53

``` df.head(5) ```

Question 54

``` df = spark.createDataFrame([(2, "Alice"), (5, "Bob")], schema=["age", "name"]) ``` Select column "name" and a new column "age" with `age_value + 10`.

Answer 54

``` df.select('name', (df.age+10).alias('age')) ``` ## Footnote To select all columns, use `df.select('*')`

Question 55

Return type of `df.show(5)` is _ whereas return type of `df.head(5)` is _ .

Answer 55

1. NoneType 2. List[Row]

Question 56

Return the datatypes of columns of `df` dataframe in comma-separated list.

Answer 56

``` df.dtypes ``` ## Footnote Each element of list is a 2-tuple: (name, type); both strings.

Question 57

Show summary of a dataset, i.e., count, mean, stddev, min, max

Answer 57

``` df.describe().show() ```

Question 58

Add new column to ``` df = spark.createDataFrame([(2, "Alice"), (5, "Bob")], schema=["age", "name"]) ``` which shows future_age after 5 years.

Answer 58

Option 1: ``` df.select("name", "age", (df["age"]+5).alias("age_after_5yrs")) ``` Option 2: ``` df.withColumn("age_after_5yrs", df["age"] + 10) ``` ## Footnote Option 2 is better if you need all columns + a new column.

Question 59

Remove column `age_after_5yrs` from `df`.

Answer 59

``` df.drop("age_after_5yrs") ```

Question 60

Remove all rows from `df` having one or more `null` entries. ## Footnote one or more = any

Answer 60

`df.dropna()`

Question 61

Remove all rows from `df` having one or more `null` entries in columns age and name. ## Footnote one or more = any

Answer 61

``` df.dropna(subset=['age', 'name']) ```

Question 62

Remove all rows from `df` having more than 2 `null` entries. ## Footnote If N is total number of columns, we don't want to allow the count of non-null columns to drop beyond N-2.

Answer 62

``` df.dropna(thresh=len(df.columns)-2) ```

Question 63

Fill all null entries in column "age" with 25.

Answer 63

``` df.fillna(25, subset='age') ```

Question 64

Fill all null entries in column "age" with 25 and in column "name" with "Unknown".

Answer 64

``` df.fillna({'age': 25, 'name': 'Unknown'}) ```

Question 65

Return rows in `df` with `age > 15`

Answer 65

Option 1 ``` df.filter(df['age'] > 15) ``` Option 2 ``` df.filter('age > 15') ```

Question 66

Return rows in `df` with `age > 15` and `name` is either 'Alice' or 'Bob'

Answer 66

Option 1 ``` df.filter((df['age'] > 15) & (df['name'].isin('Alice', 'Bob'))) ``` Option 2 ``` df.filter("age > 15 AND name IN ('Alice', 'Bob')") ```

Question 67

Return rows in `df` where `name` is neither 'Alice' nor 'Bob'.

Answer 67

``` df.filter(~df['name'].isin('Alice', 'Bob')) ```

Question 68

Given `df` has 3 columns: `name`, `age`, `score` Write pyspark code equivalent to ``` SELECT AVG(age), AVG(order) FROM df; ```

Answer 68

Option 1 ``` import pyspark.sql.functions.* df.select(avg('age').alias('avg_age'), avg('order').alias('avg_order')).show() ``` Option 2 ``` import pyspark.sql.functions.* df.groupby().agg(avg('age'), sum('order')) ``` Option 3 ``` import pyspark.sql.functions.* df.groupby().agg({'age': 'avg', 'order': 'sum'}) ```

Question 69

``` df = spark.createDataFrame([(14, "Tom", 94), (23, "Alice", 82), (23, "Alice", 102), (16, "Bob", 90)], ["age", "name", "order"]) ``` Find name of person with highest order, using PySpark code. ## Footnote SELECT name FROM df ORDER BY order DESC LIMIT 1;

Answer 69

``` import pyspark.sql.functions.* df.orderBy('order', ascending=False).select('name').limit(1).show() ```

Question 70

``` df = spark.createDataFrame([(14, "Tom", 94), (23, "Alice", 82), (23, "Alice", 102), (16, "Bob", 90)], ["age", "name", "order"]) ``` Find average order of each person (name) in `df`, using PySpark code. ## Footnote select name, avg(order) from df group by name

Answer 70

``` df.groupBy('name').agg( avg('order').alias('avg_order') ).show() ```

Question 71

``` df = spark.createDataFrame([(14, "Tom", 94), (23, "Alice", 82), (23, "Alice", 102), (16, "Bob", 90)], ["age", "name", "order"]) ``` Find total number of orders made by each person (name) in `df`, using PySpark code.

Answer 71

``` df.groupBy('name').agg( count('*').alias('order_count') ).show() ```