Databricks-Certified-Associate-Developer-for-Apache-Spark-3.0 Databricks Certified Associate Developer for Apache Spark 3.0 Exam Questions and Answers

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(1).write.option("sep", "\t").option("nullValue", "n/a").csv(csvPath)

It is important here to understand that the QUESTION NO: specifically asks for writing the DataFrame as a single CSV file. This should trigger you to think about partitions. By default, every partition

is written as a separate file, so you need to include repatition(1) into your call. coalesce(1) works here, too!

Secondly, the QUESTION NO: is very much an invitation to search through the parameters in the Spark documentation that work with DataFrameWriter.csv (link below). You will also need to know

that you need an option() statement to apply these parameters.

The final concern is about the general call structure. Once you have called accessed write of your DataFrame, options follow and then you write the DataFrame with csv. Instead of csv(csvPath),

you could also use save(csvPath, format='csv') here.

More info: pyspark.sql.DataFrameWriter.csv — PySpark 3.1.1 documentation

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

Explanation

Correct code block:

find_most_freq_letter_udf = udf(find_most_frequent_letter)

itemsDf.withColumn("most_frequent_letter", find_most_freq_letter_udf("itemName"))

Spark should use the previously registered find_most_freq_letter_udf method here – but it is not doing that in the original codeblock. There, it just uses the non-UDF version of the Python method.

Note that typically, we would have to specify a return type for udf(). Except in this case, since the default return type for udf() is a string which is what we are expecting here. If we wanted to return

an integer variable instead, we would have to register the Python function as UDF using find_most_freq_letter_udf = udf(find_most_freq_letter, IntegerType()).

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

Explanation

spark.udf.register("LIMIT_FCN", to_limit)

spark.sql("SELECT transactionId, LIMIT_FCN(predError) AS result FROM transactionsDf")

Correct! First, you have to register to_limit as UDF to use it in a sql statement. Then, you can use it under the LIMIT_FCN name, correctly naming the resulting column result.

spark.udf.register(to_limit, "LIMIT_FCN")

spark.sql("SELECT transactionId, LIMIT_FCN(predError) AS result FROM transactionsDf")

No. In this answer, the arguments to spark.udf.register are flipped.

spark.udf.register("LIMIT_FCN", to_limit)

spark.sql("SELECT transactionId, to_limit(predError) AS result FROM transactionsDf")

Wrong, this answer does not use the registered LIMIT_FCN in the sql statement, but tries to access the to_limit method directly. This will fail, since Spark cannot access it.

spark.sql("SELECT transactionId, udf(to_limit(predError)) AS result FROM transactionsDf")

Incorrect, there is no udf method in Spark's SQL.

spark.udf.register("LIMIT_FCN", to_limit)

spark.sql("SELECT transactionId, LIMIT_FCN(predError) FROM transactionsDf AS result")

False. In this answer, the column that results from applying the UDF is not correctly renamed to result.

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

Explanation

If an action including an accumulator fails during execution and Spark manages to restart the action and complete it successfully, only the successful attempt will be counted in the accumulator.

Correct, when Spark tries to rerun a failed action that includes an accumulator, it will only update the accumulator if the action succeeded.

Accumulators are immutable.

No. Although accumulators behave like write-only variables towards the executors and can only be read by the driver, they are not immutable.

All accumulators used in a Spark application are listed in the Spark UI.

Incorrect. For scala, only named, but not unnamed, accumulators are listed in the Spark UI. For pySpark, no accumulators are listed in the Spark UI – this feature is not yet implemented.

Accumulators are used to pass around lookup tables across the cluster.

Wrong – this is what broadcast variables do.

Accumulators can be instantiated directly via the accumulator(n) method of the pyspark.RDD module.

Wrong, accumulators are instantiated via the accumulator(n) method of the sparkContext, for example: counter = spark.sparkContext.accumulator(0).

More info: python - In Spark, RDDs are immutable, then how Accumulators are implemented? - Stack Overflow, apache spark - When are accumulators truly reliable? - Stack Overflow, Spark – The

Definitive Guide, Chapter 14

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)

Explanation

Correct code block:

transactionsDf.write.format("parquet").partitionBy("storeId").save("/FileStore/transactions_split")

More info: partition by - Reading files which are written using PartitionBy or BucketBy in Spark - Stack Overflow

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

Explanation

Although the QUESTION NO: suggests using a method called unique() here, that method does not actually exist in PySpark. In PySpark, it is called distinct(). But then, this method is not the right

one to use here, since with distinct() we could filter out unique values in a specific column.

However, we want to return the entire rows here. So the trick is to use dropDuplicates with the subset keyword parameter. In the documentation for dropDuplicates, the examples show that subset

should be used with a list. And this is exactly the key to solving this question: The productId column needs to be fed into the subset argument in a list, even though it is just a single column.

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

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

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)

Explanation

transactionsDf.select('value').union(transactionsDf.select('productId')).distinct()

Correct. This code block uses a common pattern for finding the unique values across multiple columns: union and distinct. In fact, it is so common that it is even mentioned in the Spark

documentation for the union command (link below).

transactionsDf.select('value', 'productId').distinct()

Wrong. This code block returns unique rows, but not unique values.

transactionsDf.agg({'value': 'collect_set', 'productId': 'collect_set'})

Incorrect. This code block will output a one-row, two-column DataFrame where each cell has an array of unique values in the respective column (even omitting any nulls).

transactionsDf.select(col('value'), col('productId')).agg({'*': 'count'})

No. This command will count the number of rows, but will not return unique values.

transactionsDf.select('value').join(transactionsDf.select('productId'), col('value')==col('productId'), 'outer')

Wrong. This command will perform an outer join of the value and productId columns. As such, it will return a two-column DataFrame. If you picked this answer, it might be a good idea for you to read

up on the difference between union and join, a link is posted below.

More info: pyspark.sql.DataFrame.union — PySpark 3.1.2 documentation, sql - What is the difference between JOIN and UNION? - Stack Overflow

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

Explanation

The challenge in this QUESTION NO: comes from there being an array variable in the schema. In addition, you should know how to pass a schema to the DataFrameReader that is invoked by

spark.read.

The correct way to define an array of strings in a schema is through ArrayType(StringType()). A schema can be passed to the DataFrameReader by simply appending schema(structType) to the

read() operator. Alternatively, you can also define a schema as a string. For example, for the schema of itemsDf, the following string would make sense: itemId integer, attributes array,

supplier string.

A thing to keep in mind is that in schema definitions, you always need to instantiate the types, like so: StringType(). Just using StringType does not work in pySpark and will fail.

Another concern with schemas is whether columns should be nullable, so allowed to have null values. In the case at hand, this is not a concern however, since the QUESTION NO: just asks for a

"valid"

schema. Both non-nullable and nullable column schemas would be valid here, since no null value appears in the DataFrame sample.

More info: Learning Spark, 2nd Edition, Chapter 3

Static notebook | Dynamic notebook: See test 3, 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)

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.

In cluster mode, the driver runs on the master node, while in client mode, the driver runs on a virtual machine in the cloud.

This is wrong, since execution modes do not specify whether workloads are run in the cloud or on-premise.

In cluster mode, each node will launch its own executor, while in client mode, executors will exclusively run on the client machine.

Wrong, since in both cases executors run on worker nodes.

In cluster mode, the driver runs on the edge node, while the client mode runs the driver in a worker node.

Wrong – in cluster mode, the driver runs on a worker node. In client mode, the driver runs on the client machine.

In client mode, the cluster manager runs on the same host as the driver, while in cluster mode, the cluster manager runs on a separate node.

No. In both modes, the cluster manager is typically on a separate node – not on the same host as the driver. It only runs on the same host as the driver in local execution mode.

More info: Learning Spark, 2nd Edition, Chapter 1, and Spark: The Definitive Guide, Chapter 15. ()

Explanation

transactionsDf.select("storeId").dropDuplicates().count()

Correct! After dropping all duplicates from column storeId, the remaining rows get counted, representing the number of unique values in the column.

transactionsDf.select(count("storeId")).dropDuplicates()

No. transactionsDf.select(count("storeId")) just returns a single-row DataFrame showing the number of non-null rows. dropDuplicates() does not have any effect in this context.

transactionsDf.dropDuplicates().agg(count("storeId"))

Incorrect. While transactionsDf.dropDuplicates() removes duplicate rows from transactionsDf, it does not do so taking only column storeId into consideration, but eliminates full row duplicates

instead.

transactionsDf.distinct().select("storeId").count()

Wrong. transactionsDf.distinct() identifies unique rows across all columns, but not only unique rows with respect to column storeId. This may leave duplicate values in the column, making the count

not represent the number of unique values in that column.

transactionsDf.select(distinct("storeId")).count()

False. There is no distinct method in pyspark.sql.functions.

Explanation

todayTransactionsDf.union(yesterdayTransactionsDf)

Correct. The union command appends rows of yesterdayTransactionsDf to the rows of todayTransactionsDf, ignoring that both DataFrames have different column names. The resulting DataFrame

will have the column names of DataFrame todayTransactionsDf.

todayTransactionsDf.unionByName(yesterdayTransactionsDf)

No. unionByName specifically tries to match columns in the two DataFrames by name and only appends values in columns with identical names across the two DataFrames. In the form presented

above, the command is a great fit for joining DataFrames that have exactly the same columns, but in a different order. In this case though, the command will fail because the two DataFrames have

different columns.

todayTransactionsDf.unionByName(yesterdayTransactionsDf, allowMissingColumns=True)

No. The unionByName command is described in the previous explanation. However, with the allowMissingColumns argument set to True, it is no longer an issue that the two DataFrames have

different column names. Any columns that do not have a match in the other DataFrame will be filled with null where there is no value. In the case at hand, the resulting DataFrame will have 7 or more

columns though, so it this command is not the right answer.

union(todayTransactionsDf, yesterdayTransactionsDf)

No, there is no union method in pyspark.sql.functions.

todayTransactionsDf.concat(yesterdayTransactionsDf)

Wrong, the DataFrame class does not have a concat method.

More info: pyspark.sql.DataFrame.union — PySpark 3.1.2 documentation, pyspark.sql.DataFrame.unionByName — PySpark 3.1.2 documentation

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

Explanation

In Spark's execution hierarchy, a job may reach over multiple stage boundaries.

Correct. A job is a sequence of stages, and thus may reach over multiple stage boundaries.

In Spark's execution hierarchy, tasks are one layer above slots.

Incorrect. Slots are not a part of the execution hierarchy. Tasks are the lowest layer.

In Spark's execution hierarchy, a stage comprises multiple jobs.

No. It is the other way around – a job consists of one or multiple stages.

In Spark's execution hierarchy, executors are the smallest unit.

False. Executors are not a part of the execution hierarchy. Tasks are the smallest unit!

In Spark's execution hierarchy, manifests are one layer above jobs.

Wrong. Manifests are not a part of the Spark ecosystem.

Explanation

Output of correct code block:

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

|itemId|attributes |supplier |

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

|1 |[winter, cozy, blue] |Sports Company Inc.|

|2 |[summer, red, fresh, cooling]|YetiX |

|3 |[travel, summer, green] |Sports Company Inc.|

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

It can be confusing to differentiate between the different sorting functions in PySpark. In this case, a particularity about sort_array has to be considered: The sort direction is given by the second

argument, not by the desc method. Luckily, this is documented in the documentation (link below). Also, for solving this QUESTION NO: you need to understand the difference between sort and

sort_array. With sort, you cannot sort values in arrays. Also, sort is a method of DataFrame, while sort_array is a method of pyspark.sql.functions.

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

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

Explanation

Correct code block:

spark.conf.set("spark.sql.shuffle.partitions", 20)

The code block expresses the option incorrectly.

Correct! The option should be expressed as a string.

The code block sets the wrong option.

No, spark.sql.shuffle.partitions is the correct option for the use case in the question.

The code block sets the incorrect number of parts.

Wrong, the code block correctly states 20 parts.

The code block uses the wrong command for setting an option.

No, in PySpark spark.conf.set() is the correct command for setting an option.

The code block is missing a parameter.

Incorrect, spark.conf.set() takes two parameters.

More info: Configuration - Spark 3.1.2 Documentation

Explanation

itemsDf.sample(fraction=0.1, seed=87238)

Correct. If itemsDf has 10,000 rows, this code block returns about 1,000, since DataFrame.sample() is never guaranteed to return an exact amount of rows. To ensure you are not returning

duplicates, you should leave the withReplacement parameter at False, which is the default. Since the QUESTION NO: specifies that the same rows should be returned even if the code block is run

twice,

you need to specify a seed. The number passed in the seed does not matter as long as it is an integer.

itemsDf.sample(withReplacement=True, fraction=0.1, seed=23536)

Incorrect. While this code block fulfills almost all requirements, it may return duplicates. This is because withReplacement is set to True.

Here is how to understand what replacement means: Imagine you have a bucket of 10,000 numbered balls and you need to take 1,000 balls at random from the bucket (similar to the problem in the

question). Now, if you would take those balls with replacement, you would take a ball, note its number, and put it back into the bucket, meaning the next time you take a ball from the bucket there

would be a chance you could take the exact same ball again. If you took the balls without replacement, you would leave the ball outside the bucket and not put it back in as you take the next 999

balls.

itemsDf.sample(fraction=1000, seed=98263)

Wrong. The fraction parameter needs to have a value between 0 and 1. In this case, it should be 0.1, since 1,000/10,000 = 0.1.

itemsDf.sampleBy("row", fractions={0: 0.1}, seed=82371)

No, DataFrame.sampleBy() is meant for stratified sampling. This means that based on the values in a column in a DataFrame, you can draw a certain fraction of rows containing those values from

the DataFrame (more details linked below). In the scenario at hand, sampleBy is not the right operator to use because you do not have any information about any column that the sampling should

depend on.

itemsDf.sample(fraction=0.1)

Incorrect. This code block checks all the boxes except that it does not ensure that when you run it a second time, the exact same rows will be returned. In order to achieve this, you would have to

specify a seed.

More info:

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

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

- Types of Samplings in PySpark 3. The explanations of the sampling… | by Pinar Ersoy | Towards Data Science

Explanation

spark.read.json(filePath)

Correct. spark.read accesses Spark's DataFrameReader. Then, Spark identifies the file type to be read as JSON type by passing filePath into the DataFrameReader.json() method.

spark.read.path(filePath)

Incorrect. Spark's DataFrameReader does not have a path method. A universal way to read in files is provided by the DataFrameReader.load() method (link below).

spark.read.path(filePath, source="json")

Wrong. A DataFrameReader.path() method does not exist (see above).

spark.read().json(filePath)

Incorrect. spark.read is a way to access Spark's DataFrameReader. However, the DataFrameReader is not callable, so calling it via spark.read() will fail.

spark.read().path(filePath)

No, Spark's DataFrameReader is not callable (see above).

More info: pyspark.sql.DataFrameReader.json — PySpark 3.1.2 documentation, pyspark.sql.DataFrameReader.load — PySpark 3.1.2 documentation

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

Explanation

Correct code block:

transactionsDf.filter(col('predError').isin([3, 6])).count()

The isin method is the correct one to use here – the in method does not exist for the Column object.

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

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

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)