Databricks-Certified-Associate-Developer-for-Apache-Spark-3.0 Sample Questions Answers

Explanation

The key here is to understand that columns that are passed to DataFrame.drop() are ignored if they do not exist in the DataFrame. So, passing column name associateId to transactionsDf.drop()

does not have any effect.

Passing a list to transactionsDf.drop() is not valid. The documentation (link below) shows the call structure as DataFrame.drop(*cols). The * means that all arguments that are passed to

DataFrame.drop() are read as columns. However, since a list of columns, for example ["predError", "productId", "associateId"] is not a column, Spark will run into an error.

More info: pyspark.sql.DataFrame.drop — PySpark 3.1.1 documentation

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 50 (Databricks import instructions)

Explanation

Correct code block:

transactionsDf.select(col("storeId").cast(StringType()))

Solving this QUESTION NO: involves understanding that, when using types from the pyspark.sql.types such as StringType, these types need to be instantiated when using them in Spark, or, in

simple words, they need to be followed by parentheses like so: StringType(). You could also use .cast("string") instead, but that option is not given here.

More info: pyspark.sql.Column.cast — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 37 (Databricks import instructions)

Explanation

The storage level is inappropriate for fault-tolerant storage.

Correct. Typically, when thinking about fault tolerance and storage levels, you would want to store redundant copies of the dataset. This can be achieved by using a storage level such as

StorageLevel.MEMORY_AND_DISK_2.

The code block uses the wrong command for caching.

Wrong. In this case, DataFrame.persist() needs to be used, since this operator supports passing a storage level. DataFrame.cache() does not support passing a storage level.

Caching is not supported in Spark, data are always recomputed.

Incorrect. Caching is an important component of Spark, since it can help to accelerate Spark programs to great extent. Caching is often a good idea for datasets that need to be accessed

repeatedly.

Data caching capabilities can be accessed through the spark object, but not through the DataFrame API.

No. Caching is either accessed through DataFrame.cache() or DataFrame.persist().

The DataFrameWriter needs to be invoked.

Wrong. The DataFrameWriter can be accessed via DataFrame.write and is used to write data to external data stores, mostly on disk. Here, we find keywords such as "cache" and "executor

memory" that point us away from using external data stores. We aim to save data to memory to accelerate the reading process, since reading from disk is comparatively slower. The

DataFrameWriter does not write to memory, so we cannot use it here.

More info: Best practices for caching in Spark SQL | by David Vrba | Towards Data Science

Explanation

Correct code block:

transactionsDf.withColumn("transactionDateForm", from_unixtime("transactionDate", "MMM d (EEEE)"))

The QUESTION NO: specifically asks about "adding" a column. In the context of all presented answers, DataFrame.withColumn() is the correct command for this. In theory, DataFrame.select() could

also be

used for this purpose, if all existing columns are selected and a new one is added. DataFrame.withColumnRenamed() is not the appropriate command, since it can only rename existing columns, but

cannot add a new column or change the value of a column.

Once DataFrame.withColumn() is chosen, you can read in the documentation (see below) that the first input argument to the method should be the column name of the new column.

The final difficulty is the date format. The QUESTION NO: indicates that the date format Apr 26 (Sunday) is desired. The answers give "MMM d (EEEE)" and "MM d (EEE)" as options. It can be hard

to

know the details of the date format that is used in Spark. Specifically, knowing the differences between MMM and MM is probably not something you deal with every day. But, there is an easy way

to remember the difference: M (one letter) is usually the shortest form: 4 for April. MM includes padding: 04 for April. MMM (three letters) is the three-letter month abbreviation: Apr for April. And

MMMM is the longest possible form: April. Knowing this four-letter sequence helps you select the correct option here.

More info: pyspark.sql.DataFrame.withColumn — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 3, QUESTION NO: 35 (Databricks import instructions)

There is a place in the Spark UI that shows the property spark.executor.memory.

Correct, you can see Spark properties such as spark.executor.memory in the Environment tab.

Some of the tabs in the Spark UI are named Jobs, Stages, Storage, DAGs, Executors, and SQL.

Wrong – Jobs, Stages, Storage, Executors, and SQL are all tabs in the Spark UI. DAGs can be inspected in the "Jobs" tab in the job details or in the Stages or SQL tab, but are not a separate tab.

Via the Spark UI, workloads can be manually distributed across distributors.

No, the Spark UI is meant for inspecting the inner workings of Spark which ultimately helps understand, debug, and optimize Spark transactions.

Via the Spark UI, stage execution speed can be modified.

No, see above.

The Scheduler tab shows how jobs that are run in parallel by multiple users are distributed across the cluster.

No, there is no Scheduler tab.

Explanation

transactionsDf.groupBy("productId").count()

Correct. This code block first groups DataFrame transactionsDf by column productId and then counts the rows in each group.

transactionsDf.groupBy("productId").select(count("value"))

Incorrect. You cannot call select on a GroupedData object (the output of a groupBy) statement.

transactionsDf.count("productId")

No. DataFrame.count() does not take any arguments.

transactionsDf.count("productId").distinct()

Wrong. Since DataFrame.count() does not take any arguments, this option cannot be right.

transactionsDf.groupBy("productId").agg(col("value").count())

False. A Column object, as returned by col("value"), does not have a count() method. You can see all available methods for Column object linked in the Spark documentation below.

More info: pyspark.sql.DataFrame.count — PySpark 3.1.2 documentation, pyspark.sql.Column — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 3, QUESTION NO: 41 (Databricks import instructions)

While only one of these code blocks works, the DataFrame API is pretty flexible when it comes to accepting columns into the pow() method. The following code blocks would also work:

transactionsDf.withColumn("predErrorSquared", pow("predError", 2))

transactionsDf.withColumn("predErrorSquared", pow("predError", lit(2)))

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 26 (Databricks import instructions) (https://flrs.github.io/spark_practice_tests_code/#1/26.html ,

https://bit.ly/sparkpracticeexams_import_instructions)

Explanation

Correct code block:

transactionsDf = spark.read.load("data/transactions.csv", sep=";", format="csv", header=True, inferSchema=True)

By default, Spark does not infer the schema of the CSV (since this usually takes some time). So, you need to add the inferSchema=True option to the code block.

More info: pyspark.sql.DataFrameReader.csv — PySpark 3.1.2 documentation

Explanation

Output of correct code block:

+-------------------+

| supplier|

+-------------------+

|Sports Company Inc.|

+-------------------+

Key to managing this QUESTION NO: is understand which operator to use to do the opposite of an operation – the ~ (not) operator. In addition, you should know that there is no unique() method.

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 55 (Databricks import instructions)

Explanation

DataFrame.collect() sends all data in a DataFrame from executors to the driver, so this generally causes a great amount of network traffic in comparison to the other options listed.

DataFrame.coalesce() just reduces the number of partitions and generally aims to reduce network traffic in comparison to a full shuffle.

DataFrame.select() is evaluated lazily and, unless followed by an action, does not cause significant network traffic.

DataFrame.rdd.map() is evaluated lazily, it does therefore not cause great amounts of network traffic.

DataFrame.count() is an action. While it does cause some network traffic, for the same DataFrame, collecting all data in the driver would generally be considered to cause a greater amount of

network traffic.

Explanation

The driver receives data upon request by actions.

Correct! Actions trigger the distributed execution of tasks on executors which, upon task completion, transfer result data back to the driver.

Actions are Spark's way of exchanging data between executors.

No. In Spark, data is exchanged between executors via shuffles.

Writing data to disk is the primary purpose of actions.

No. The primary purpose of actions is to access data that is stored in Spark's RDDs and return the data, often in aggregated form, back to the driver.

Actions are Spark's way of modifying RDDs.

Incorrect. Firstly, RDDs are immutable – they cannot be modified. Secondly, Spark generates new RDDs via transformations and not actions.

Stage boundaries are commonly established by actions.

Wrong. A stage boundary is commonly established by a shuffle, for example caused by a wide transformation.

Explanation

Correct code block:

itemsDf.withColumn("itemNameElements", split("itemName"," "))

Output of code block:

+------+----------------------------------+-------------------+------------------------------------------+

|itemId|itemName |supplier |itemNameElements |

+------+----------------------------------+-------------------+------------------------------------------+

|1 |Thick Coat for Walking in the Snow|Sports Company Inc.|[Thick, Coat, for, Walking, in, the, Snow]|

|2 |Elegant Outdoors Summer Dress |YetiX |[Elegant, Outdoors, Summer, Dress] |

|3 |Outdoors Backpack |Sports Company Inc.|[Outdoors, Backpack] |

+------+----------------------------------+-------------------+------------------------------------------+

The key to solving this QUESTION NO: is that the split method definitely needs a second argument here (also look at the link to the documentation below). Given the values in column itemName in

DataFrame itemsDf, this should be a space character " ". This is the character we need to split the words in the column.

More info: pyspark.sql.functions.split — PySpark 3.1.1 documentation

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 46 (Databricks import instructions)

Explanation

2,3

Correct! Both panels 2 and 3 represent the operation performed for broadcast variables. While a broadcast operation may look like panel 3, with the driver being the bottleneck, it most probably

looks like panel 2.

This is because the torrent protocol sits behind Spark's broadcast implementation. In the torrent protocol, each executor will try to fetch missing broadcast variables from the driver or other nodes,

preventing the driver from being the bottleneck.

1,2

Wrong. While panel 2 may represent broadcasting, panel 1 shows bi-directional communication which does not occur in broadcast operations.

3

No. While broadcasting may materialize like shown in panel 3, its use of the torrent protocol also enables communciation as shown in panel 2 (see first explanation).

1,3,4

No. While panel 2 shows broadcasting, panel 1 shows bi-directional communication – not a characteristic of broadcasting. Panel 4 shows uni-directional communication, but in the wrong direction.

Panel 4 resembles more an accumulator variable than a broadcast variable.

2,5

Incorrect. While panel 2 shows broadcasting, panel 5 includes bi-directional communication – not a characteristic of broadcasting.

More info: Broadcast Join with Spark – henning.kropponline.de

Explanation

Correct code block:

transactionsDf.write.format("parquet").mode("overwrite").option("compression", "snappy").save(storeDir)

Solving this QUESTION NO: requires you to know how to access the DataFrameWriter (link below) from the DataFrame API - through DataFrame.write.

Another nuance here is about knowing the different modes available for writing parquet files that determine Spark's behavior when dealing with existing files. These, together with the compression

options are explained in the DataFrameWriter.parquet documentation linked below.

Finally, bracket __5__ poses a certain challenge. You need to know which command you can use to pass down the file path to the DataFrameWriter. Both save and parquet are valid options here.

More info:

- DataFrame.write: pyspark.sql.DataFrame.write — PySpark 3.1.1 documentation

- DataFrameWriter.parquet: pyspark.sql.DataFrameWriter.parquet — PySpark 3.1.1 documentation

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 58 (Databricks import instructions)

Explanation

The output is the typical output of a DataFrame.printSchema() call. The DataFrame's RDD representation does not have a printSchema or formatSchema method (find available methods in the RDD

documentation linked below). The output of print(transactionsDf.schema) is this: StructType(List(StructField(transactionId,IntegerType,true),StructField(predError,IntegerType,true),StructField

(value,IntegerType,true),StructField(storeId,IntegerType,true),StructField(productId,IntegerType,true),StructField(f,IntegerType,true))). It includes the same information as the nicely formatted original

output, but is not nicely formatted itself. Lastly, the DataFrame's schema attribute does not have a print() method.

More info:

- pyspark.RDD: pyspark.RDD — PySpark 3.1.2 documentation

- DataFrame.printSchema(): pyspark.sql.DataFrame.printSchema — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 52 (Databricks import instructions)

Explanation

Spark's DataFrames are equal to Python's or R's DataFrames.

No, they are not equal. They are only similar. A major difference between Spark and Python is that Spark's DataFrames are distributed, whereby Python's are not.

Explanation

Explanation

Spark provides a way to digest JSON-formatted strings as schema. However, it is not trivial to use. Although slightly above exam difficulty, this QUESTION NO: is beneficial to your exam

preparation, since

it helps you to familiarize yourself with the concept of enforcing schemas on data you are reading in - a topic within the scope of the exam.

The first answer that jumps out here is the one that uses spark.read.schema instead of spark.read.json. Looking at the documentation of spark.read.schema (linked below), we notice that the

operator expects types pyspark.sql.types.StructType or str as its first argument. While variable json_schema is a string, the documentation states that the str should be "a DDL-formatted string (For

example col0 INT, col1 DOUBLE)". Variable json_schema does not contain a string in this type of format, so this answer option must be wrong.

With four potentially correct answers to go, we now look at the schema parameter of spark.read.json() (documentation linked below). Here, too, the schema parameter expects an input of type

pyspark.sql.types.StructType or "a DDL-formatted string (For example col0 INT, col1 DOUBLE)". We already know that json_schema does not follow this format, so we should focus on how we can

transform json_schema into pyspark.sql.types.StructType. Hereby, we also eliminate the option where schema=json_schema.

The option that includes schema=spark.read.json(json_schema) is also a wrong pick, since spark.read.json returns a DataFrame, and not a pyspark.sql.types.StructType type.

Ruling out the option which includes schema_of_json(json_schema) is rather difficult. The operator's documentation (linked below) states that it "[p]arses a JSON string and infers its schema in DDL

format". This use case is slightly different from the case at hand: json_schema already is a schema definition, it does not make sense to "infer" a schema from it. In the documentation you can see

an example use case which helps you understand the difference better. Here, you pass string '{a: 1}' to schema_of_json() and the method infers a DDL-format schema STRUCT from it.

In our case, we may end up with the output schema of schema_of_json() describing the schema of the JSON schema, instead of using the schema itself. This is not the right answer option.

Now you may consider looking at the StructType.fromJson() method. It returns a variable of type StructType - exactly the type which the schema parameter of spark.read.json expects.

Although we could have looked at the correct answer option earlier, this explanation is kept as exhaustive as necessary to teach you how to systematically eliminate wrong answer options.

More info:

- pyspark.sql.DataFrameReader.schema — PySpark 3.1.2 documentation

- pyspark.sql.DataFrameReader.json — PySpark 3.1.2 documentation

- pyspark.sql.functions.schema_of_json — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 3, QUESTION NO: 51 (Databricks import instructions)

Explanation

The key to solving this QUESTION NO: is knowing (or reading in the documentation) that, by default, cache() stores values to memory and writes any partitions for which there is insufficient memory

to disk. persist() can achieve the exact same behavior, however not with the StorageLevel.MEMORY_ONLY option listed here. It is also worth noting that cache() does not have any arguments.

If you have troubles finding the storage level information in the documentation, please also see this student Q&A thread that sheds some light here.

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 30 (Databricks import instructions)

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 23 (Databricks import instructions) (https://flrs.github.io/spark_practice_tests_code/#1/23.html ,

https://bit.ly/sparkpracticeexams_import_instructions)

Explanation

Use a narrow transformation to reduce the number of partitions.

Correct! DataFrame.coalesce(n) is a narrow transformation, and in fact the most efficient way to resize the DataFrame of all options listed. One would run DataFrame.coalesce(8) to resize the

DataFrame.

Use operation DataFrame.coalesce(8) to fully shuffle the DataFrame and reduce the number of partitions.

Wrong. The coalesce operation avoids a full shuffle, but will shuffle data if needed. This answer is incorrect because it says "fully shuffle" – this is something the coalesce operation will not do. As a

general rule, it will reduce the number of partitions with the very least movement of data possible. More info: distributed computing - Spark - repartition() vs coalesce() - Stack Overflow

Use operation DataFrame.coalesce(0.5) to halve the number of partitions in the DataFrame.

Incorrect, since the num_partitions parameter needs to be an integer number defining the exact number of partitions desired after the operation. More info: pyspark.sql.DataFrame.coalesce —

PySpark 3.1.2 documentation

Use operation DataFrame.repartition(8) to shuffle the DataFrame and reduce the number of partitions.

No. The repartition operation will fully shuffle the DataFrame. This is not the most efficient way of reducing the number of partitions of all listed options.

Use a wide transformation to reduce the number of partitions.

No. While possible via the DataFrame.repartition(n) command, the resulting full shuffle is not the most efficient way of reducing the number of partitions.

Explanation

In difficulty, this QUESTION NO: is above what you can expect from the exam. What this QUESTION NO: wants to teach you, however, is to pay attention to the useful details included in the

documentation.

pyspark.sql.corr is not a very common method, but it deals with Spark's data structure in an interesting way. The command takes two columns over multiple rows and returns a single row - similar to

an aggregation function. When examining the documentation (linked below), you will find this code example:

a = range(20)

b = [2 * x for x in range(20)]

df = spark.createDataFrame(zip(a, b), ["a", "b"])

df.agg(corr("a", "b").alias('c')).collect()

[Row(c=1.0)]

See how corr just returns a single row? Once you understand this, you should be suspicious about answers that include first(), since there is no need to just select a single row. A reason to eliminate

those answers is that DataFrame.first() returns an object of type Row, but not DataFrame, as requested in the question.

transactionsDf.select(corr(col("predError"), col("value")).alias("corr"))

Correct! After calculating the Pearson correlation coefficient, the resulting column is correctly renamed to corr.

transactionsDf.select(corr(predError, value).alias("corr"))

No. In this answer, Python will interpret column names predError and value as variable names.

transactionsDf.select(corr(col("predError"), col("value")).alias("corr")).first()

Incorrect. first() returns a row, not a DataFrame (see above and linked documentation below).

transactionsDf.select(corr("predError", "value"))

Wrong. Whie this statement returns a DataFrame in the desired shape, the column will have the name corr(predError, value) and not corr.

transactionsDf.select(corr(["predError", "value"]).alias("corr")).first()

False. In addition to first() returning a row, this code block also uses the wrong call structure for command corr which takes two arguments (the two columns to correlate).

More info:

- pyspark.sql.functions.corr — PySpark 3.1.2 documentation

- pyspark.sql.DataFrame.first — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 3, QUESTION NO: 53 (Databricks import instructions)

Explanation

Correct code block:

transactionsDf.repartition(14, "storeId", "transactionDate").count()

Since we do not know how many partitions DataFrame transactionsDf has, we cannot safely use coalesce, since it would not make any change if the current number of partitions is smaller than 14.

So, we need to use repartition.

In the Spark documentation, the call structure for repartition is shown like this: DataFrame.repartition(numPartitions, *cols). The * operator means that any argument after numPartitions will be

interpreted as column. Therefore, the brackets need to be removed.

Finally, the QUESTION NO: specifies that after the execution the DataFrame should be divided. So, indirectly this QUESTION NO: is asking us to append an action to the code block. Since .select()

is a transformation. the only possible choice here is .count().

More info: pyspark.sql.DataFrame.repartition — PySpark 3.1.1 documentation

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 40 (Databricks import instructions)

Explanation

More info: pyspark.sql.DataFrame.join — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 27 (Databricks import instructions)

transactionsDf.dropna(thresh=4)

Correct. Note that by only working with the thresh keyword argument, the first how keyword argument is ignored. Also, figuring out which value to set for thresh can be difficult, especially when

under pressure in the exam. Here, I recommend you use the notes to create a "simulation" of what different values for thresh would do to a DataFrame. Here is an explanatory image why thresh=4 is

the correct answer to the question:

transactionsDf.dropna(thresh=2)

Almost right. See the comment about thresh for the correct answer above.

transactionsDf.dropna("any")

No, this would remove all rows that have at least one missing value.

transactionsDf.drop.na("",2)

No, drop.na is not a proper DataFrame method.

transactionsDf.dropna("",4)

No, this does not work and will throw an error in Spark because Spark cannot understand the first argument.

More info: pyspark.sql.DataFrame.dropna — PySpark 3.1.1 documentation (https://bit.ly/2QZpiCp)

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 20 (Databricks import instructions) (https://flrs.github.io/spark_practice_tests_code/#1/20.html ,

https://bit.ly/sparkpracticeexams_import_instructions)

Explanation

Correct code block:

accum=sc.accumulator(0)

def check_if_inc_in_supplier(row):

if 'Inc.' in row['supplier']:

accum.add(1)

itemsDf.foreach(check_if_inc_in_supplier)

print(accum.value)

To answer this QUESTION NO: correctly, you need to know both about the DataFrame.foreach() method and accumulators.

When Spark runs the code, it executes it on the executors. The executors do not have any information about variables outside of their scope. This is whhy simply using a Python variable counter,

like in the two examples that start with counter = 0, will not work. You need to tell the executors explicitly that counter is a special shared variable, an Accumulator, which is managed by the driver

and can be accessed by all executors for the purpose of adding to it.

If you have used Pandas in the past, you might be familiar with the iterrows() command. Notice that there is no such command in PySpark.

The two examples that start with print do not work, since DataFrame.foreach() does not have a return value.

More info: pyspark.sql.DataFrame.foreach — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 3, QUESTION NO: 22 (Databricks import instructions)

Explanation

Result of correct code block:

+-------------------+

|attributes_exploded|

+-------------------+

| winter|

| cooling|

+-------------------+

To solve this question, you need to know about explode(). This operation helps you to split up arrays into single rows. If you did not have a chance to familiarize yourself with this method yet, find

more examples in the documentation (link below).

Note that explode() is a method made available through pyspark.sql.functions – it is not available as a method of a DataFrame or a Column, as written in some of the answer options.

More info: pyspark.sql.functions.explode — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 18 (Databricks import instructions)

Explanation

show() is the correct method to look for here, since the QUESTION NO: specifically asks for displaying the rows in a nicely formatted way. Here is the output of show (only a few rows shown):

+-------------+---------+-----+-------+---------+----+---------------+

|transactionId|predError|value|storeId|productId| f|transactionDate|

+-------------+---------+-----+-------+---------+----+---------------+

| 3| 3| 1| 25| 3|null| 1585824821|

| 5| null| 2| null| 2|null| 1575285427|

| 4| null| 3| 3| 2|null| 1583244275|

+-------------+---------+-----+-------+---------+----+---------------+

With regards to the sorting, specifically in ascending order since the smallest values should be shown first, the following expressions are valid:

- transactionsDf.sort(col("value")) ("ascending" is the default sort direction in the sort method)

- transactionsDf.sort(asc(col("value")))

- transactionsDf.sort(asc("value"))

- transactionsDf.sort(transactionsDf.value.asc())

- transactionsDf.sort(transactionsDf.value)

Also, orderBy is just an alias of sort, so all of these expressions work equally well using orderBy.

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 43 (Databricks import instructions)

Explanation

Correct code block:

from pyspark import StorageLevel

transactionsDf.persist(StorageLevel.MEMORY_ONLY_2).count()

Only persist takes different storage levels, so any option using cache() cannot be correct. persist() is evaluated lazily, so an action needs to follow this command. select() is not an action, but count()

is – so all options using select() are incorrect.

Finally, the QUESTION NO: states that "the executors' memory should be utilized as much as possible, but not writing anything to disk". This points to a MEMORY_ONLY storage level. In this

storage level, partitions that do not fit into memory will be recomputed when they are needed, instead of being written to disk, as with the storage option MEMORY_AND_DISK. Since the data need

to be duplicated across two executors, _2 needs to be appended to the storage level.

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 25 (Databricks import instructions)

Explanation

Correct code block:

transactionsDf.withColumnRenamed("storeId", "storeNumber")

More info: pyspark.sql.DataFrame.withColumnRenamed — PySpark 3.1.1 documentation

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 38 (Databricks import instructions)

Explanation

Filtering out distinct rows based on columns is achieved with the dropDuplicates method, not the distinct method which does not take any arguments.

The second argument of the join() method only accepts strings if they are column names. The SQL-like statement "itemsDf.itemId==transactionsDf.productId" is therefore invalid.

In addition, it is not necessary to specify how="inner", since the default join type for the join command is already inner.

More info: pyspark.sql.DataFrame.join — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 53 (Databricks import instructions)

Explanation

This QUESTION NO: asks for your knowledge about the cast syntax. cast is a method of the Column class. It is worth noting that one could also convert a column type using the Column.astype()

method, which is just an alias for cast.

Find more info in the documentation linked below.

More info: pyspark.sql.Column.cast — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 33 (Databricks import instructions)

Explanation

Recognizing that the UDF specification requires a return type (unless it is a string, which is the default) is important for solving this question. In addition, you should make sure that the generated

UDF (evaluateTestSuccessUDF) and not the Python function (evaluateTestSuccess) is applied to column storeId.

More info: pyspark.sql.functions.udf — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 34 (Databricks import instructions)

Explanation

The schema passed into schema should be of type StructType or a string, so all entries in which a list is passed are incorrect.

In addition, since all numbers are whole numbers, the IntegerType() data type is the correct option here. NumberType() is not a valid data type and StringType() would fail, since the parquet file is

stored in the "most appropriate format for this kind of data", meaning that it is most likely an IntegerType, and Spark does not convert data types if a schema is provided.

Also note that StructType accepts only a single argument (a list of StructFields). So, passing multiple arguments is invalid.

Finally, Spark needs to know which format the file is in. However, all of the options listed are valid here, since Spark assumes parquet as a default when no file format is specifically passed.

More info: pyspark.sql.DataFrameReader.schema — PySpark 3.1.2 documentation and StructType — PySpark 3.1.2 documentation

Explanation

In this QUESTION NO: it is important to realize that you are asked to sort transactionDf by two columns. This means that the sorting of the second column depends on the sorting of the first column.

So, any option that sorts the entire DataFrame (through chaining sort statements) will not work. The two columns need to be channeled through the same call to sort().

Also, order_by is not a valid DataFrame API method.

More info: pyspark.sql.DataFrame.sort — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 22 (Databricks import instructions)

Explanation

Caching means storing a copy of a partition on an executor, so it can be accessed quicker by subsequent operations, instead of having to be recalculated. cache() is a lazily-evaluated method of the

DataFrame. Since count() is an action (while filter() is not), it triggers the caching process.

More info: pyspark.sql.DataFrame.cache — PySpark 3.1.2 documentation, Learning Spark, 2nd Edition, Chapter 7

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 20 (Databricks import instructions)

Explanation

Answering this QUESTION NO: correctly depends on whether you understand the arguments to the DataFrame.sample() method (link to the documentation below). The arguments are as follows:

DataFrame.sample(withReplacement=None, fraction=None, seed=None).

The first argument withReplacement specified whether a row can be drawn from the DataFrame multiple times. By default, this option is disabled in Spark. But we have to enable it here, since the question asks for a row being able to appear more than once. So, we need to pass True for this argument.

About replacement: "Replacement" is easiest explained with the example of removing random items from a box. When you remove those "with replacement" it means that after you have taken an

item out of the box, you put it back inside. So, essentially, if you would randomly take 10 items out of a box with 100 items, there is a chance you take the same item twice or more times. "Without

replacement" means that you would not put the item back into the box after removing it. So, every time you remove an item from the box, there is one less item in the box and you can never take the

same item twice.

The second argument to the withReplacement method is fraction. This referes to the fraction of items that should be returned. In the QUESTION NO: we are asked for 150 out of 1000 items – a

fraction of 0.15.

The last argument is a random seed. A random seed makes a randomized processed repeatable. This means that if you would re-run the same sample() operation with the same random seed, you

would get the same rows returned from the sample() command. There is no behavior around the random seed specified in the question. The varying random seeds are only there to confuse you!

More info: pyspark.sql.DataFrame.sample — PySpark 3.1.1 documentation

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 49 (Databricks import instructions)

Explanation

A narrow transformation is an operation in which no data is exchanged across the cluster.

Correct! In narrow transformations, no data is exchanged across the cluster, since these transformations do not require any data from outside of the partition they are applied on. Typical narrow

transformations include filter, drop, and coalesce.

A narrow transformation is an operation in which data is exchanged across partitions.

No, that would be one definition of a wide transformation, but not of a narrow transformation. Wide transformations typically cause a shuffle, in which data is exchanged across partitions, executors,

and the cluster.

A narrow transformation is an operation in which data is exchanged across the cluster.

No, see explanation just above this one.

A narrow transformation is a process in which 32-bit float variables are cast to smaller float variables, like 16-bit or 8-bit float variables.

No, type conversion has nothing to do with narrow transformations in Spark.

A narrow transformation is a process in which data from multiple RDDs is used.

No. A resilient distributed dataset (RDD) can be described as a collection of partitions. In a narrow transformation, no data is exchanged between partitions. Thus, no data is exchanged between

RDDs.

One could say though that a narrow transformation and, in fact, any transformation results in a new RDD being created. This is because a transformation results in a change to an existing RDD

(RDDs are the foundation of other Spark data structures, like DataFrames). But, since RDDs are immutable, a new RDD needs to be created to reflect the change caused by the transformation.

More info: Spark Transformation and Action: A Deep Dive | by Misbah Uddin | CodeX | Medium

Explanation

Execution is triggered by transformations.

Correct. Execution is triggered by actions only, not by transformations.

Lineages allow Spark to coalesce transformations into stages.

Incorrect. In Spark, lineage means a recording of transformations. This lineage enables lazy evaluation in Spark.

Predicate pushdown is a feature resulting from lazy evaluation.

Wrong. Predicate pushdown means that, for example, Spark will execute filters as early in the process as possible so that it deals with the least possible amount of data in subsequent

transformations, resulting in a performance improvements.

Accumulators do not change the lazy evaluation model of Spark.

Incorrect. In Spark, accumulators are only updated when the query that refers to the is actually executed. In other words, they are not updated if the query is not (yet) executed due to lazy

evaluation.

Spark will fail a job only during execution, but not during definition.

Wrong. During definition, due to lazy evaluation, the job is not executed and thus certain errors, for example reading from a non-existing file, cannot be caught. To be caught, the job needs to be

executed, for example through an action.

Explanation

distinct() is a method of a DataFrame. Knowing this, or recognizing this from the documentation, is the key to solving this question.

More info: pyspark.sql.DataFrame.distinct — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 19 (Databricks import instructions)

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 25 (Databricks import instructions) (https://flrs.github.io/spark_practice_tests_code/#1/25.html ,

https://bit.ly/sparkpracticeexams_import_instructions)

Explanation

Correct code block:

transactionsDf.withColumn("cos", round(cos(degrees(transactionsDf.value)),2))

This QUESTION NO: is especially confusing because col, "cos" are so similar. Similar-looking answer options can also appear in the exam and, just like in this question, you need to pay attention to

the

details to identify what the correct answer option is.

The first answer option to throw out is the one that starts with withColumnRenamed: The QUESTION NO: speaks specifically of adding a column. The withColumnRenamed operator only renames

an

existing column, however, so you cannot use it here.

Next, you will have to decide what should be in gap 2, the first argument of transactionsDf.withColumn(). Looking at the documentation (linked below), you can find out that the first argument of

withColumn actually needs to be a string with the name of the column to be added. So, any answer that includes col("cos") as the option for gap 2 can be disregarded.

This leaves you with two possible answers. The real difference between these two answers is where the cos and degree methods are, either in gaps 3 and 4, or vice-versa. From the QUESTION

NO: you

can find out that the new column should have "the values in column value converted to degrees and having the cosine of those converted values taken". This prescribes you a clear order of

operations: First, you convert values from column value to degrees and then you take the cosine of those values. So, the inner parenthesis (gap 4) should contain the degree method and then,

logically, gap 3 holds the cos method. This leaves you with just one possible correct answer.

More info: pyspark.sql.DataFrame.withColumn — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 3, QUESTION NO: 49 (Databricks import instructions)

Explanation

Tasks get assigned to the executors by the driver.

Correct! Or, in other words: Executors take the tasks that they were assigned to by the driver, run them over partitions, and report the their outcomes back to the driver.

Tasks transform jobs into DAGs.

No, this statement disrespects the order of elements in the Spark hierarchy. The Spark driver transforms jobs into DAGs. Each job consists of one or more stages. Each stage contains one or more

tasks.

A task is a collection of rows.

Wrong. A partition is a collection of rows. Tasks have little to do with a collection of rows. If anything, a task processes a specific partition.

A task is a command sent from the driver to the executors in response to a transformation.

Incorrect. The Spark driver does not send anything to the executors in response to a transformation, since transformations are evaluated lazily. So, the Spark driver would send tasks to executors

only in response to actions.

A task is a collection of slots.

No. Executors have one or more slots to process tasks and each slot can be assigned a task.

Explanation

To solve this question, you need to know that DataFrame.sample() is not guaranteed to return the exact fraction of the number of rows specified as an argument. Furthermore, since duplicates may

be returned, you should understand that the operator's withReplacement argument should be set to True. A force= argument for the operator does not exist.

While the take argument returns an exact number of rows, it will just take the first specified number of rows (1000 in this question) from the DataFrame. Since the DataFrame does not include

duplicate rows, there is no potential of any of those returned rows being duplicates when using take(), so the correct answer cannot involve take().

More info: pyspark.sql.DataFrame.sample — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 41 (Databricks import instructions)

Explanation

Executors stop upon application completion by default.

Correct. Executors only persist during the lifetime of an application.

A notable exception to that is when Dynamic Resource Allocation is enabled (which it is not by default). With Dynamic Resource Allocation enabled, executors are terminated when they are idle,

independent of whether the application has been completed or not.

An executor can serve multiple applications.

Wrong. An executor is always specific to the application. It is terminated when the application completes (exception see above).

Each node hosts a single executor.

No. Each node can host one or more executors.

Executors store data in memory only.

No. Executors can store data in memory or on disk.

Executors are launched by the driver.

Incorrect. Executors are launched by the cluster manager on behalf of the driver.

More info: Job Scheduling - Spark 3.1.2 Documentation, How Applications are Executed on a Spark Cluster | Anatomy of a Spark Application | InformIT, and Spark Jargon for Starters. This blog is to

clear some of the… | by Mageswaran D | Medium

Explanation

Correct code block:

from pyspark.sql.functions import udf

from pyspark.sql import types as T

transactionsDf.createOrReplaceTempView('transactions')

def pow_5(x):

if x:

return x**5

return x

spark.udf.register('power_5_udf', pow_5, T.LongType())

spark.sql('SELECT power_5_udf(value) AS result FROM transactions')

Here it is important to understand how the pow_5 method handles empty values. In the wrong code block above, the pow_5 method is unable to handle empty values and will throw an error, since

Python's ** operator cannot deal with any null value Spark passes into method pow_5.

The order of arguments for registering the UDF function with Spark via spark.udf.register matters. In the code snippet in the question, the arguments for the SQL method name and the actual Python

function are switched. You can read more about the arguments of spark.udf.register and see some examples of its usage in the documentation (link below).

Finally, you should recognize that in the original code block, an expression to rename column created through the UDF function is missing. The renaming is done by SQL's AS result argument.

Omitting that argument, you end up with the column name power_5_udf(value) and not result.

More info: pyspark.sql.functions.udf — PySpark 3.1.1 documentation

Explanation

itemsDf.write.mode("overwrite").parquet(filePath)

Correct! itemsDf.write returns a pyspark.sql.DataFrameWriter instance whose overwriting behavior can be modified via the mode setting or by passing mode="overwrite" to the parquet() command.

Although the parquet format is not prescribed for solving this question, parquet() is a valid operator to initiate Spark to write the data to disk.

itemsDf.write.mode("overwrite").path(filePath)

No. A pyspark.sql.DataFrameWriter instance does not have a path() method.

itemsDf.write.option("parquet").mode("overwrite").path(filePath)

Incorrect, see above. In addition, a file format cannot be passed via the option() method.

itemsDf.write(filePath, mode="overwrite")

Wrong. Unfortunately, this is too simple. You need to obtain access to a DataFrameWriter for the DataFrame through calling itemsDf.write upon which you can apply further methods to control how

Spark data should be written to disk. You cannot, however, pass arguments to itemsDf.write directly.

itemsDf.write().parquet(filePath, mode="overwrite")

False. See above.

More info: pyspark.sql.DataFrameWriter.parquet — PySpark 3.1.2 documentation

Static notebook | Dynamic notebook: See test 3, QUESTION NO: 56 (Databricks import instructions)

Explanation

As a general rule: After having gone through the practice tests you probably have a good feeling for what classifies as a wide and what classifies as a narrow transformation. If you are unsure, feel

free to play around in Spark and display the explanation of the Spark execution plan via DataFrame.[operation, for example sort()].explain(). If repartitioning is involved, it would count as a wide

transformation.

DataFrame.select()

Correct! A wide transformation includes a shuffle, meaning that an input partition maps to one or more output partitions. This is expensive and causes traffic across the cluster. With the select()

operation however, you pass commands to Spark that tell Spark to perform an operation on a specific slice of any partition. For this, Spark does not need to exchange data across partitions, each

partition can be worked on independently. Thus, you do not cause a wide transformation.

DataFrame.repartition()

Incorrect. When you repartition a DataFrame, you redefine partition boundaries. Data will flow across your cluster and end up in different partitions after the repartitioning is completed. This is

known as a shuffle and, in turn, is classified as a wide transformation.

DataFrame.aggregate()

No. When you aggregate, you may compare and summarize data across partitions. In the process, data are exchanged across the cluster, and newly formed output partitions depend on one or more

input partitions. This is a typical characteristic of a shuffle, meaning that the aggregate operation may classify as a wide transformation.

DataFrame.join()

Wrong. Joining multiple DataFrames usually means that large amounts of data are exchanged across the cluster, as new partitions are formed. This is a shuffle and therefore DataFrame.join()

counts as a wide transformation.

DataFrame.sort()

False. When sorting, Spark needs to compare many rows across all partitions to each other. This is an expensive operation, since data is exchanged across the cluster and new partitions are

formed as data is reordered. This process classifies as a shuffle and, as a result, DataFrame.sort() counts as wide transformation.

More info: Understanding Apache Spark Shuffle | Philipp Brunenberg

Explanation:

The correct code block looks like this:

Then, the first couple of rows of itemAttributesDf look like this:

C:\Users\Admin\Desktop\Data\Odt data\Untitled.jpg

explode() is not a method of DataFrame. explode() should be used inside the select() method instead.

This is correct.

The split() method should be used inside the select() method instead of the explode() method.

No, the split() method is used to split strings into parts. However, column attributs is an array of strings. In this case, the explode() method is appropriate.

Since itemId is the index, it does not need to be an argument to the select() method.

No, itemId still needs to be selected, whether it is used as an index or not.

The explode() method expects a Column object rather than a string.

No, a string works just fine here. This being said, there are some valid alternatives to passing in a string:

The alias() method needs to be called after the select() method.

No.

More info: pyspark.sql.functions.explode — PySpark 3.1.1 documentation (https://bit.ly/2QUZI1J)

Static notebook | Dynamic notebook: See test 1, QUESTION NO: 22 (Databricks import instructions) (https://flrs.github.io/spark_practice_tests_code/#1/22.html , https://bit.ly/sparkpracticeexams_import_instructions)

The hierarchy is, from top to bottom: Job, Stage, Task.

Executors and slots facilitate the execution of tasks, but they are not directly part of the hierarchy. Executors are launched by the driver on worker nodes for the purpose of running a specific Spark

application. Slots help Spark parallelize work. An executor can have multiple slots which enable it to process multiple tasks in parallel.