Correct syntax is;
df.orderBy(asc_nulls_last(“a“))
df.sort(asc_nulls_last(“a“))

When you write a text file, you need to be sure to have only one string column; otherwise, the write will fail:

Read the questions and responses carefully ! You will have many questions like this one, try to visualise it and write it down if it helps. There is always quotes in the column name and you need to you .cast to cast a column

Spark SQL (including SQL and the DataFrame and Dataset APIs) does not guarantee the order of evaluation of subexpressions. In particular, the inputs of an operator or function are not necessarily evaluated left-to-right or in any other fixed order. For example, logical AND and OR expressions do not have left-to-right “short-circuiting” semantics. To perform proper null checking, we recommend that you do either of the following: Make the UDF itself null-aware and do null checking inside the UDF itself Use IF or CASE WHEN expressions to do the null check and invoke the UDF in a conditional branch

Here is the explication;
1) df.groupBy().sum()
This part is resulting type DataFrame[sum(Number): bigint]
If we show the resulting dataframe it would be;
+———–+
|sum(Number)|
+———–+
| 130|
+———–+
2) df.groupBy().sum().collect()
We do a collect on the previous summed dataframe; Remember collect returns a list of rows; (https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrame.collect.html#pyspark.sql.DataFrame.collect)
Here is the resulting list of row.
[Row(sum(Number)=130)]
3) df.groupBy().sum().collect()[0]
We have a list of rows, and we are only interested in the first row object which will give;
Row(sum(Number)=130)
4) df.groupBy().sum().collect()[0][0]
We have our row, but we are interested getting the value of it. For our example it is ‘130‘.
The fields in rows can be accessed:
a) like attributes (row.key)
b) like dictionary values (row[key])
In this example, we chose to access it as a dictionary value, hence the second [0]
(https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.Row.html#pyspark.sql.Row)
Or another way to do it is; import pyspark.sql.functions as F        df.agg(F.sum(“Number“)).collect()[0][0]

Familiarize yourself with the syntax of withColumn and withColumnRenamed.

The highest level abstraction in Spark SQL is the Catalog. The Catalog is an abstraction for the storage of metadata about the data stored in your tables as well as other helpful things like databases, tables, functions, and views

First set of optimizations takes place in step logical optimization. See the link for more detail: https://databricks.com/glossary/catalyst-optimizer

One important note is the concept of managed versus unmanaged tables. Tables store two important pieces of information. The data within the tables as well as the data about the tables; that is, the metadata. You can have Spark manage the metadata for a set of files as well as for the data. When you define a table from files on disk, you are defining an unmanaged table. When you use saveAsTable on a DataFrame, you are creating a managed table for which Spark will track of all of the relevant information.

Global temp views are accessed via prefix ‘global_temp’ And other tables are accessed without any prefixes.

At each stage boundary, data is written to disk by tasks in the parent stages and then fetched over the network by tasks in the child stage. Because they incur heavy disk and network I/O, stage boundaries can be expensive and should be avoided when possible.

If you don‘t specify number of partitions normally, spark tries to set the number of partitions automatically based on your cluster but here we specified that we want to have 8 partitions after we created the dataframe.

The second argument for DataFrame.withColumn should be a Column so you have to use a literal to add constant value “1”:

You need to query the right table. Read carefully the questions !

In cluster mode, a user submits a pre-compiled JAR, Python script, or R script to a cluster manager. The cluster manager then launches the driver process on a worker node inside the cluster, in addition to the executor processes.

You need to be really familiar with the syntax of withColumn, withColumnRenamed for the exam. Learn them very well.

Review well this concept for the exam. https://spark.apache.org/docs/latest/cluster-overview.html

see https://spark.apache.org/docs/latest/sql-performance-tuning.html

With Spark 2.0+ we can use CSV connector to read a tsv file.

correct syntax is;
df.repartition(8)
and you cannot increase the number of partitions with
df.coalesce(8)

If you are dropping an unmanaged table, no data will be removed but you will no longer be able to refer to this data by the table name.

A worker node can be holding multiple executors (processes) if it has sufficient CPU, Memory and Storage. https://spark.apache.org/docs/latest/cluster-overview.html

Client mode is nearly the same as cluster mode except that the Spark driver remains on the client machine that submitted the application.

Correct function is: to_date and also we need to give a format which is compatible with java timestamp class https://docs.oracle.com/javase/8/docs/api/java/sql/Timestamp.html

Correct syntax is schema.add(“new_column“,StringType(),True).

If you would like to run multiple Spark Applications on the same cluster, Spark provides a mechanism to dynamically adjust the resources your application occupies based on the workload. This means that your application can give resources back to the cluster if they are no longer used, and request them again later when there is demand. This feature is particularly useful if multiple applications share resources in your Spark cluster. This feature is disabled by default and available on all coarse-grained cluster managers; that is, standalone mode, YARN mode, and Mesos coarse-grained mode. There are two requirements for using this feature. First, your application must set spark.dynamicAllocation.enabled to true. Second, you must set up an external shuffle service on each worker node in the same cluster and set spark.shuffle.service.enabled to true in your application. The purpose of the external shuffle service is to allow executors to be removed without deleting shuffle files written by them. The Spark Fair Scheduler specifies resource pools and allocates jobs to different resource pools to achieve resource scheduling within an application. In this way, the computing resources are effectively used and the runtime of jobs is balanced, ensuring that the subsequently-submitted jobs are not affected by over-loaded jobs.

We control the parallelism of files that we write by controlling the partitions prior to writing and therefore the number of partitions before writing equals to number of files created after the write operation. If you don‘t specify number of partitions normally, spark tries to set the number of partitions automatically based on your cluster but here we specified that we want to have 8 partitions after we created the dataframe.

It is preferable to process in parallel big data sets distributed across a cluster with spark

dense_rank() window function is used to get the result with rank of rows within a window partition without any gaps. This is similar to rank() function difference being rank function leaves gaps in rank when there are ties.

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.

see https://spark.apache.org/docs/latest/rdd-programming-guide.html#transformations

All transformations are lazily evaluated in spark.

Default storage level is MEMORY_AND_DISK
https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrame.cache.html#pyspark.sql.DataFrame.cache
https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html
Dataframe has unpersist() function;
https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.sql.DataFrame.unpersist.html#pyspark.sql.DataFrame.unpersist
The persist() method needs an action to load data from its source to materialize the DataFrame in cache (and you should be using all the partitions while doing that action)

Correct answer here is select one column, we just select one column and get a new dataframe from it

Correct usage is spark.catalog.uncacheTable(“tableName”)
To remove the data from the cache, just call:
spark.sql(“uncache table table_name“)
or
spark.catalog.uncacheTable(“table_name”)
To unpersist dataframe use:
spark.sql(“df.unpersist()“)
Another thing to remember is when using DataFrame.persist() data on disk is always serialized.

An execution mode gives you the power to determine where the aforementioned resources are physically located when you go to run your application. You have three modes to choose from: Cluster mode, client mode and local mode. Standalone is one of the cluster manager types.

see https://spark.apache.org/docs/latest/api/java/index.html?org/apache/spark/storage/StorageLevel.html

DPP can auto-optimize your queries and make them more performant automatically. For more information; https://spark.apache.org/docs/latest/configuration.html#dynamic-allocation

Correct useage of drop is the following:

Parquet is the default file format. If you don’t include the format() method, the DataFrame will still be saved as a Parquet file. If we don’t include mode ‘overwrite’ our application will crash since there is already a file exist with the same name.

The driver is the machine in which the application runs. It is responsible for three main things: 1) Maintaining information about the Spark Application, 2) Responding to the user’s program, 3) Analyzing, distributing, and scheduling work across the executors.

see https://spark.apache.org/docs/latest/rdd-programming-guide.html#actions

Correct syntax is
df.na.drop(“any“)
Example
import org.apache.spark.sql.functions._
val rawData = Seq((“A“, “20“),(“B“, “30“),(“C“, null))
val df = spark.createDataFrame(rawData).toDF(“Letter“, “Number“)
val df2 = df.na.drop(“any“)
df2.show()
Result
+——+——+ |
Letter|Number|
+——+——+
| A| 20|
| B| 30|
+——+——+

see https://spark.apache.org/docs/latest/rdd-programming-guide.html#transformations

see the following blog for more information: https://databricks.com/blog/2020/05/29/adaptive-query-execution-speeding-up-spark-sql-at-runtime.html

You need to name the accumulator in order to see in it in the spark ui For accumulator restarted tasks will not update the value in case of a failure. In transformations, each task’s update can be applied more than once if tasks or job stages are re-executed. Accumulators provide a mutable variable that a Spark cluster can safely update on a per-row basis. You can define your own custom accumulator class by extending org.apache.spark.util.AccumulatorV2 in Java or Scala or pyspark.AccumulatorParam in Python.

Correct answer is joinType = “right_outer“. For example df1.join(person, joinExpression, “right_outer”).show()

Broadcast variables are a way you can share an immutable value efficiently around the cluster without encapsulating that variable in a function closure. The normal way to use a variable in your driver node inside your tasks is to simply reference it in your function closures (e.g., in a map operation), but this can be inefficient, especially for large variables such as a lookup table or a machine learning model. The reason for this is that when you use a variable in a closure, it must be deserialized on the worker nodes many times (one per task)

An executor is a Java Virtual Machine (JVM) running on a worker node. See the componenets here: https://spark.apache.org/docs/latest/cluster-overview.html

When you use cache() or persist(), the DataFrame is not fully cached until you invoke an action that goes through every record (e.g., count()). If you use an action like take(1), only one partition will be cached because Catalyst realizes that you do not need to compute all the partitions just to retrieve one record.

JVM garbage collection can be a problem when you have large “churn” in terms of the RDDs stored by your program. When Java needs to evict old objects to make room for new ones, it will need to trace through all your Java objects and find the unused ones. The main point to remember here is that the cost of garbage collection is proportional to the number of Java objects, so using data structures with fewer objects (e.g. an array of Ints instead of a LinkedList) greatly lowers this cost. https://spark.apache.org/docs/latest/tuning.html#garbage-collection-tuning

Date_sub and date_add are some functions that exist in the following packages org.apache.spark.sql.functions.*

df1.join(df2, joinExpression, joinType)

Coalse function avoids a full shuffle if it‘s known that the number is decreasing then the executor can safely keep data on the minimum number of partitions, only moving the data off the extra nodes, onto the nodes that we kept. And it cannot be used to increase the number of partitions.

There are other syntax’s also but to broadcast dfA but for this example you need to wrap it in keyword broadcast. Also the order of the join is important as you can see.

https://spark.apache.org/docs/latest/cluster-overview.html