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

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)

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

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

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

Correct code block:

spark.createDataFrame([( " red " ,), ( " blue " ,), ( " green " ,)], [ " color " ])

The createDataFrame syntax is not exactly straightforward, but luckily the documentation (linked below) provides several examples on how to use it. It also shows an example very similar to the

code block presented here which should help you answer this QUESTION NO: correctly.

More info: pyspark.sql.SparkSession.createDataFrame — PySpark 3.1.2 documentation

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

Explanation

Correct code block:

transactionsDf.select((col( " storeId " ).between(20, 30)) & (col( " productId " )==2))

Although this QUESTION NO: may make you think that it asks for a filter or where statement, it does not. It asks explicity to return a column with booleans – this should point you to the select

statement.

Another trick here is the rarely used between() method. It exists and resolves to ((storeId > = 20) AND (storeId < = 30)) in SQL. geq() and leq() do not exist.

Another riddle here is how to chain the two conditions. The only valid answer here is & . Operators like & & or and are not valid. Other boolean operators that would be valid in Spark are | and.

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

Explanation

MEMORY_AND_DISK replicates cached DataFrames both on memory and disk.

Correct, this statement is wrong. Spark prioritizes storage in memory, and will only store data on disk that does not fit into memory.

DISK_ONLY will not use the worker node ' s memory.

Wrong, this statement is correct. DISK_ONLY keeps data only on the worker node ' s disk, but not in memory.

In client mode, DataFrames cached with the MEMORY_ONLY_2 level will not be stored in the edge node ' s memory.

Wrong, this statement is correct. In fact, Spark does not have a provision to cache DataFrames in the driver (which sits on the edge node in client mode). Spark caches DataFrames in the executors '

memory.

Caching can be undone using the DataFrame.unpersist() operator.

Wrong, this statement is correct. Caching, as achieved via the DataFrame.cache() or DataFrame.persist() operators can be undone using the DataFrame.unpersist() operator. This operator will

remove all of its parts from the executors ' memory and disk.

The cache operator on DataFrames is evaluated like a transformation.

Wrong, this statement is correct. DataFrame.cache() is evaluated like a transformation: Through lazy evaluation. This means that after calling DataFrame.cache() the command will not have any

effect until you call a subsequent action, like DataFrame.cache().count().

More info: pyspark.sql.DataFrame.unpersist — PySpark 3.1.2 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

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

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

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

Explanation

Correct code block:

importedDf = spark.read.json(jsonPath)

importedDf.createOrReplaceTempView( " importedDf " )

spark.sql( " SELECT * FROM importedDf WHERE productId != 3 " )

Option 5 is the only correct way listed of reading in a JSON in PySpark. The option( " format " , " json " ) is not the correct way to tell Spark ' s DataFrameReader that you want to read a JSON file. You

would do this through format( " json " ) instead. Also, you can communicate the specific path of the JSON file to the DataFramReader using the load() method, not the path() method.

In order to use a SQL command through the SparkSession spark, you first need to create a temporary view through DataFrame.createOrReplaceTempView().

The SQL statement should start with the SELECT operator. The FILTER operator SQL provides is not the correct one to use here.

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

Explanation

Correct code block:

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

The conf interface is part of the SparkSession, so you need to call it through spark and not pyspark. To configure spark, you need to use the set method, not the get method. get reads a property,

but does not write it. The correct property to achieve what is outlined in the QUESTION NO: is spark.sql.aggregate.partitions, which needs to be passed to set as a string. Properties

spark.shuffle.partitions and spark.sql.aggregate.partitions do not exist in Spark.

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

Explanation

Correct code block:

transactionsDf.select([ " transactionId " , " predError " , " value " , " f " ])

The DataFrame.select returns specific columns from the DataFrame and accepts a list as its only argument. Thus, this is the correct choice here. The option using col([ " transactionId " , " predError " ,

" value " , " f " ]) is invalid, since inside col(), one can only pass a single column name, not a list. Likewise, all columns being specified in a single string like " transactionId, predError, value, f " is not valid

syntax.

filter and where filter rows based on conditions, they do not control which columns to return.

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 49 (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)

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.join(broadcast(itemsDf), " transactionId " , " left_semi " )

This QUESTION NO: is extremely difficult and exceeds the difficulty of questions in the exam by far.

A first indication of what is asked from you here is the remark that " the query should be executed in an optimized way " . You also have qualitative information about the size of itemsDf and

transactionsDf. Given that itemsDf is " very small " and that the execution should be optimized, you should consider instructing Spark to perform a broadcast join, broadcasting the " very small "

DataFrame itemsDf to all executors. You can explicitly suggest this to Spark via wrapping itemsDf into a broadcast() operator. One answer option does not include this operator, so you can disregard

it. Another answer option wraps the broadcast() operator around transactionsDf - the bigger of the two DataFrames. This answer option does not make sense in the optimization context and can

likewise be disregarded.

When thinking about the broadcast() operator, you may also remember that it is a method of pyspark.sql.functions. One answer option, however, resolves to itemsDf.broadcast([...]). The DataFrame

class has no broadcast() method, so this answer option can be eliminated as well.

All two remaining answer options resolve to transactionsDf.join([...]) in the first 2 gaps, so you will have to figure out the details of the join now. You can pick between an outer and a left semi join. An

outer join would include columns from both DataFrames, where a left semi join only includes columns from the " left " table, here transactionsDf, just as asked for by the question. So, the correct

answer is the one that uses the left_semi join.

Explanation

Correct code block:

transactionsDf.select( " storeId " ).printSchema()

The difficulty of this QUESTION NO: is that it is hard to solve with the stepwise first-to-last-gap approach that has worked well for similar questions, since the answer options are so different from

one

another. Instead, you might want to eliminate answers by looking for patterns of frequently wrong answers.

A first pattern that you may recognize by now is that column names are not expressed in quotes. For this reason, the answer that includes storeId should be eliminated.

By now, you may have understood that the DataFrame.limit() is useful for returning a specified amount of rows. It has nothing to do with specific columns. For this reason, the answer that resolves to

limit( " storeId " ) can be eliminated.

Given that we are interested in information about the data type, you should QUESTION NO: whether the answer that resolves to limit(1).columns provides you with this information. While

DataFrame.columns is a valid call, it will only report back column names, but not column types. So, you can eliminate this option.

The two remaining options either use the printSchema() or print_schema() command. You may remember that DataFrame.printSchema() is the only valid command of the two. The select( " storeId " )

part just returns the storeId column of transactionsDf - this works here, since we are only interested in that column ' s type anyways.

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

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

Explanation

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

Explanation

Output of correct code block:

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

|itemName |col |

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

|Thick Coat for Walking in the Snow|blue |

|Thick Coat for Walking in the Snow|winter|

|Thick Coat for Walking in the Snow|cozy |

|Outdoors Backpack |green |

|Outdoors Backpack |summer|

|Outdoors Backpack |travel|

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

The key to solving this QUESTION NO: is knowing about Spark ' s explode operator. Using this operator, you can extract values from arrays into single rows. The following guidance steps through

the

answers systematically from the first to the last gap. Note that there are many ways to solving the gap questions and filtering out wrong answers, you do not always have to start filtering out from the

first gap, but can also exclude some answers based on obvious problems you see with them.

The answers to the first gap present you with two options: filter and where. These two are actually synonyms in PySpark, so using either of those is fine. The answer options to this gap therefore do

not help us in selecting the right answer.

The second gap is more interesting. One answer option includes " Sports " .isin(col( " Supplier " )). This construct does not work, since Python ' s string does not have an isin method. Another option

contains col(supplier). Here, Python will try to interpret supplier as a variable. We have not set this variable, so this is not a viable answer. Then, you are left with answers options that include col

( " supplier " ).contains( " Sports " ) and col( " supplier " ).isin( " Sports " ). The QUESTION NO: states that we are looking for suppliers whose name includes Sports, so we have to go for the contains operator

here.

We would use the isin operator if we wanted to filter out for supplier names that match any entries in a list of supplier names.

Finally, we are left with two answers that fill the third gap both with " itemName " and the fourth gap either with explode( " attributes " ) or " attributes " . While both are correct Spark syntax, only explode

( " attributes " ) will help us achieve our goal. Specifically, the QUESTION NO: asks for one attribute from column attributes per row - this is what the explode() operator does.

One answer option also includes array_explode() which is not a valid operator in PySpark.

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

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

Manually persisting RDDs in Spark prevents them from being garbage collected.

This statement is incorrect, and thus the correct answer to the question. Spark ' s garbage collector will remove even persisted objects, albeit in an " LRU " fashion. LRU stands for least recently used.

So, during a garbage collection run, the objects that were used the longest time ago will be garbage collected first.

See the linked StackOverflow post below for more information.

Serialized caching is a strategy to increase the performance of garbage collection.

This statement is correct. The more Java objects Spark needs to collect during garbage collection, the longer it takes. Storing a collection of many Java objects, such as a DataFrame with a

complex schema, through serialization as a single byte array thus increases performance. This means that garbage collection takes less time on a serialized DataFrame than an unserialized

DataFrame.

Optimizing garbage collection performance in Spark may limit caching ability.

This statement is correct. A full garbage collection run slows down a Spark application. When taking about " tuning " garbage collection, we mean reducing the amount or duration of these slowdowns.

A full garbage collection run is triggered when the Old generation of the Java heap space is almost full. (If you are unfamiliar with this concept, check out the link to the Garbage Collection Tuning docs below.) Thus, one measure to avoid triggering a garbage collection run is to prevent the Old generation share of the heap space to be almost full.

To achieve this, one may decrease its size. Objects with sizes greater than the Old generation space will then be discarded instead of cached (stored) in the space and helping it to be " almost full " .

This will decrease the number of full garbage collection runs, increasing overall performance.

Inevitably, however, objects will need to be recomputed when they are needed. So, this mechanism only works when a Spark application needs to reuse cached data as little as possible.

Garbage collection information can be accessed in the Spark UI ' s stage detail view.

This statement is correct. The task table in the Spark UI ' s stage detail view has a " GC Time " column, indicating the garbage collection time needed per task.

In Spark, using the G1 garbage collector is an alternative to using the default Parallel garbage collector.

This statement is correct. The G1 garbage collector, also known as garbage first garbage collector, is an alternative to the default Parallel garbage collector.

While the default Parallel garbage collector divides the heap into a few static regions, the G1 garbage collector divides the heap into many small regions that are created dynamically. The G1

garbage collector has certain advantages over the Parallel garbage collector which improve performance particularly for Spark workloads that require high throughput and low latency.

The G1 garbage collector is not enabled by default, and you need to explicitly pass an argument to Spark to enable it. For more information about the two garbage collectors, check out the

Databricks article linked below.

Explanation

In a shuffle, Spark writes data to disk.

Correct! Spark ' s architecture dictates that intermediate results during a shuffle are written to disk.

A shuffle is one of many actions in Spark.

Incorrect. A shuffle is a transformation, but not an action.

Shuffles involve only single partitions.

No, shuffles involve multiple partitions. During a shuffle, Spark generates output partitions from multiple input partitions.

Operations involving shuffles are never evaluated lazily.

Wrong. A shuffle is a costly operation and Spark will evaluate it as lazily as other transformations. This is, until a subsequent action triggers its evaluation.

Shuffles belong to a class known as " full transformations " .

Not quite. Shuffles belong to a class known as " wide transformations " . " Full transformation " is not a relevant term in Spark.

More info: Spark – The Definitive Guide, Chapter 2 and Spark: disk I/O on stage boundaries explanation - Stack Overflow

Explanation

To mitigate skew, Spark automatically disregards null values in keys when joining.

This statement is incorrect, and thus the correct answer to the question. Joining keys that contain null values is of particular concern with regard to data skew.

In real-world applications, a table may contain a great number of records that do not have a value assigned to the column used as a join key. During the join, the data is at risk of being heavily

skewed. This is because all records with a null-value join key are then evaluated as a single large partition, standing in stark contrast to the potentially diverse key values (and therefore small

partitions) of the non-null-key records.

Spark specifically does not handle this automatically. However, there are several strategies to mitigate this problem like discarding null values temporarily, only to merge them back later (see last link

below).

In skewed DataFrames, the largest and the smallest partition consume very different amounts of memory.

This statement is correct. In fact, having very different partition sizes is the very definition of skew. Skew can degrade Spark performance because the largest partition occupies a single executor for

a long time. This blocks a Spark job and is an inefficient use of resources, since other executors that processed smaller partitions need to idle until the large partition is processed.

Salting can resolve data skew.

This statement is correct. The purpose of salting is to provide Spark with an opportunity to repartition data into partitions of similar size, based on a salted partitioning key.

A salted partitioning key typically is a column that consists of uniformly distributed random numbers. The number of unique entries in the partitioning key column should match the number of your

desired number of partitions. After repartitioning by the salted key, all partitions should have roughly the same size.

Spark does not automatically optimize skew joins by default.

This statement is correct. Automatic skew join optimization is a feature of Adaptive Query Execution (AQE). By default, AQE is disabled in Spark. To enable it, Spark ' s spark.sql.adaptive.enabled

configuration option needs to be set to true instead of leaving it at the default false.

To automatically optimize skew joins, Spark ' s spark.sql.adaptive.skewJoin.enabled options also needs to be set to true, which it is by default.

When skew join optimization is enabled, Spark recognizes skew joins and optimizes them by splitting the bigger partitions into smaller partitions which leads to performance increases.

Broadcast joins are a viable way to increase join performance for skewed data over sort-merge joins.

This statement is correct. Broadcast joins can indeed help increase join performance for skewed data, under some conditions. One of the DataFrames to be joined needs to be small enough to fit

into each executor ' s memory, along a partition from the other DataFrame. If this is the case, a broadcast join increases join performance over a sort-merge join.

The reason is that a sort-merge join with skewed data involves excessive shuffling. During shuffling, data is sent around the cluster, ultimately slowing down the Spark application. For skewed data,

the amount of data, and thus the slowdown, is particularly big.

Broadcast joins, however, help reduce shuffling data. The smaller table is directly stored on all executors, eliminating a great amount of network traffic, ultimately increasing join performance relative

to the sort-merge join.

It is worth noting that for optimizing skew join behavior it may make sense to manually adjust Spark ' s spark.sql.autoBroadcastJoinThreshold configuration property if the smaller DataFrame is bigger

than the 10 MB set by default.

More info:

- Performance Tuning - Spark 3.0.0 Documentation

- Data Skew and Garbage Collection to Improve Spark Performance

- Section 1.2 - Joins on Skewed Data • GitBook

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

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)

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

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)

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:

transactionsDf.withColumn( ' associateId ' , lit(5)).drop( ' productId ' , ' value ' )

For solving this QUESTION NO: it is important that you know the lit() function (link to documentation below). This function enables you to add a column of a constant value to a DataFrame.

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

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

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.filter(col( " productId " ) > = 2).limit(2)

The filter and where operators in gap 1 are just aliases of one another, so you cannot use them to pick the right answer.

The column definition in gap 2 is more helpful. The DataFrame.filter() method takes an argument of type Column or str. From all possible answers, only the one including col( " productId " ) > = 2 fits

this profile, since it returns a Column type.

The answer option using " productId " > 2 is invalid, since Spark does not understand that " productId " refers to column productId. The answer option using transactionsDf[productId] > = 2 is wrong

because you cannot refer to a column using square bracket notation in Spark (if you are coming from Python using Pandas, this is something to watch out for). In all other options, productId is being

referred to as a Python variable, so they are relatively easy to eliminate.

Also note that the QUESTION NO: asks for the value in column productId being at least 2. This translates to a " greater or equal " sign ( > = 2), but not a " greater " sign ( > 2).

Another thing worth noting is that there is no DataFrame.max() method. If you picked any option including this, you may be confusing it with the pyspark.sql.functions.max method. The correct

method to limit the amount of rows is the DataFrame.limit() method.

More info:

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

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

Static notebook | Dynamic notebook: See test 3, QUESTION NO: 54 (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

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

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

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

This QUESTION NO: targets your knowledge about how to chain filtering conditions. Each filtering condition should be in parentheses. The correct operator for " or " is the pipe character (|) and not

the word or. Another operator of concern is the equality operator. For the purpose of comparison, equality is expressed as two equal signs (==).

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

Explanation

Correct code block:

schema = StructType([

StructField( " itemId " , IntegerType(), True),

StructField( " attributes " , ArrayType(StringType(), True), True),

StructField( " supplier " , StringType(), True)

])

spark.read.options(modifiedBefore= " 2029-03-20T05:44:46 " ).schema(schema).parquet(filePath)

This QUESTION NO: is more difficult than what you would encounter in the exam. In the exam, for this QUESTION NO: type, only one error needs to be identified and not " one or multiple " as in the

question.

Columns in the schema definition use the wrong object type, the modification date threshold is specified incorrectly, and Spark cannot identify the file format.

Correct! Columns in the schema definition should use the StructField type. Building a schema from pyspark.sql.types, as here using classes like StructType and StructField, is one of multiple ways

of expressing a schema in Spark. A StructType always contains a list of StructFields (see documentation linked below). So, nesting StructType and StructType as shown in the QUESTION NO: is

wrong.

The modification date threshold should be specified by a keyword argument like options(modifiedBefore= " 2029-03-20T05:44:46 " ) and not two consecutive non-keyword arguments as in the original

code block (see documentation linked below).

Spark cannot identify the file format correctly, because either it has to be specified by using the DataFrameReader.format(), as an argument to DataFrameReader.load(), or directly by calling, for

example, DataFrameReader.parquet().

Columns in the schema are unable to handle empty values and the modification date threshold is specified incorrectly.

No. If StructField would be used for the columns instead of StructType (see above), the third argument specified whether the column is nullable. The original schema shows that columns should be

nullable and this is specified correctly by the third argument being True in the schema in the code block.

It is correct, however, that the modification date threshold is specified incorrectly (see above).

The attributes array is specified incorrectly, Spark cannot identify the file format, and the syntax of the call to Spark ' s DataFrameReader is incorrect.

Wrong. The attributes array is specified correctly, following the syntax for ArrayType (see linked documentation below). That Spark cannot identify the file format is correct, see correct answer

above. In addition, the DataFrameReader is called correctly through the SparkSession spark.

Columns in the schema definition use the wrong object type and the syntax of the call to Spark ' s DataFrameReader is incorrect.

Incorrect, the object types in the schema definition are correct and syntax of the call to Spark ' s DataFrameReader is correct.

The data type of the schema is incompatible with the schema() operator and the modification date threshold is specified incorrectly.

False. The data type of the schema is StructType and an accepted data type for the DataFrameReader.schema() method. It is correct however that the modification date threshold is specified

incorrectly (see correct answer above).

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

Standalone mode uses only a single executor per worker per application.

This is correct and a limitation of Spark ' s standalone mode.

Standalone mode is a viable solution for clusters that run multiple frameworks.

Incorrect. A limitation of standalone mode is that Apache Spark must be the only framework running on the cluster. If you would want to run multiple frameworks on the same cluster in parallel, for

example Apache Spark and Apache Flink, you would consider the YARN deployment mode.

Standalone mode uses a single JVM to run Spark driver and executor processes.

No, this is what local mode does.

Standalone mode is how Spark runs on YARN and Mesos clusters.

No. YARN and Mesos modes are two deployment modes that are different from standalone mode. These modes allow Spark to run alongside other frameworks on a cluster. When Spark is run in

standalone mode, only the Spark framework can run on the cluster.

Standalone mode means that the cluster does not contain the driver.

Incorrect, the cluster does not contain the driver in client mode, but in standalone mode the driver runs on a node in the cluster.

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

Explanation

Pay attention here to which variables are quoted. fileSchema is a variable and thus should not be in quotes. parquet is not a variable and therefore should be in quotes.

SparkSession.read (here referenced as spark.read) returns a DataFrameReader which all subsequent calls reference - the DataFrameReader is not callable, so you should not use parentheses

here.

Finally, there is no open method in PySpark. The method name is load.

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

Explanation

DAG stands for " Directing Acyclic Graph " .

No, DAG stands for " Directed Acyclic Graph " .

Spark strategically hides DAGs from developers, since the high degree of automation in Spark means that developers never need to consider DAG layouts.

No, quite the opposite. You can access DAGs through the Spark UI and they can be of great help when optimizing queries manually.

In contrast to transformations, DAGs are never lazily executed.

DAGs represent the execution plan in Spark and as such are lazily executed when the driver requests the data processed in the DAG.

Explanation

This QUESTION NO: tests your knowledge about how to use the groupBy and agg pattern in Spark. Using the documentation, you can find out that there is no average() method in

pyspark.sql.functions.

Static notebook | Dynamic notebook: See test 2, QUESTION NO: 42 (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

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

https://bit.ly/sparkpracticeexams_import_instructions)

Explanation

transactionsDf.unpersist()

Correct. The DataFrame.unpersist() command does exactly what the QUESTION NO: asks for - it removes all cached parts of the DataFrame from memory and disk.

del transactionsDf

False. While this option can help remove the DataFrame from memory and disk, it does not do so immediately. The reason is that this command just notifies the Python garbage collector that the

transactionsDf now may be deleted from memory. However, the garbage collector does not do so immediately and, if you wanted it to run immediately, would need to be specifically triggered to do

so. Find more information linked below.

array_remove(transactionsDf, " * " )

Incorrect. The array_remove method from pyspark.sql.functions is used for removing elements from arrays in columns that match a specific condition. Also, the first argument would be a column, and

not a DataFrame as shown in the code block.

transactionsDf.persist()

No. This code block does exactly the opposite of what is asked for: It caches (writes) DataFrame transactionsDf to memory and disk. Note that even though you do not pass in a specific storage

level here, Spark will use the default storage level (MEMORY_AND_DISK).

transactionsDf.clearCache()

Wrong. Spark ' s DataFrame does not have a clearCache() method.

More info: pyspark.sql.DataFrame.unpersist — PySpark 3.1.2 documentation, python - How to delete an RDD in PySpark for the purpose of releasing resources? - Stack Overflow

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

Explanation

Correct code block:

transactionsDf.select( " storeId " , " predError " ).collect()

collect() is a method of the DataFrame object.

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

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

Explanation

Correct code block:

transactionsDf.join(itemsDf, transactionsDf.productId==itemsDf.itemId, how= ' inner ' ).filter(~isnull(col( ' value ' ))).count()

Expressions col( " transactionsDf.productId " ) and col( " itemsDf.itemId " ) are invalid. col() does not accept the name of a DataFrame, only column names.

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

Explanation

This QUESTION NO: deals with the parameters of Spark ' s split operator for strings.

To solve this question, you first need to understand the difference between DataFrame.withColumn() and DataFrame.withColumnRenamed(). The correct option here is DataFrame.withColumn()

since, according to the question, we want to add a column and not rename an existing column. This leaves you with only 3 answers to consider.

The second gap should be filled with the name of the new column to be added to the DataFrame. One of the remaining answers states the column name as itemNameBetweenSeparators, while the

other two state it as " itemNameBetweenSeparators " . The correct option here is " itemNameBetweenSeparators " , since the other option would let Python try to interpret itemNameBetweenSeparators

as the name of a variable, which we have not defined. This leaves you with 2 answers to consider.

The decision boils down to how to fill gap 5. Either with 4 or with 5. The QUESTION NO: asks for arrays of maximum four strings. The code in gap 5 relates to the limit parameter of Spark ' s split

operator

(see documentation linked below). The documentation states that " the resulting array ' s length will not be more than limit " , meaning that we should pick the answer option with 4 as the code in the

fifth gap here.

On a side note: One answer option includes a function str_split. This function does not exist in pySpark.

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

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

Dynamic partition pruning reoptimizes query plans based on runtime statistics collected during query execution.

No – this is what adaptive query execution does, but not dynamic partition pruning.

Dynamic partition pruning concatenates columns of similar data types to optimize join performance.

Wrong, this answer does not make sense, especially related to dynamic partition pruning.

Dynamic partition pruning reoptimizes physical plans based on data types and broadcast variables.

It is true that dynamic partition pruning works in joins using broadcast variables. This actually happens in both the logical optimization and the physical planning stage. However, data types do not

play a role for the reoptimization.

Dynamic partition pruning performs wide transformations on disk instead of in memory.

This answer does not make sense. Dynamic partition pruning is meant to accelerate Spark – performing any transformation involving disk instead of memory resources would decelerate Spark and

certainly achieve the opposite effect of what dynamic partition pruning is intended for.

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

Correct code block:

len(spark.read.csv(filePath, comment= ' # ' ).columns)

This is a challenging QUESTION NO: with difficulties in an unusual context: The boundary between DataFrame and the DataFrameReader. It is unlikely that a QUESTION NO: of this difficulty level

appears in the

exam. However, solving it helps you get more comfortable with the DataFrameReader, a subject you will likely have to deal with in the exam.

Before dealing with the inner parentheses, it is easier to figure out the outer parentheses, gaps 1 and 5. Given the code block, the object in gap 5 would have to be evaluated by the object in gap 1,

returning the number of columns in the read-in CSV. One answer option includes DataFrame in gap 1 and shape[0] in gap 2. Since DataFrame cannot be used to evaluate shape[0] , we can discard

this answer option.

Other answer options include size in gap 1. size() is not a built-in Python command, so if we use it, it would have to come from somewhere else. pyspark.sql.functions includes a size() method, but

this method only returns the length of an array or map stored within a column (documentation linked below). So, using a size() method is not an option here. This leaves us with two potentially valid

answers.

We have to pick between gaps 2 and 3 being spark.read or pyspark.DataFrameReader. Looking at the documentation (linked below), the DataFrameReader is actually a child class of pyspark.sql,

which means that we cannot import it using pyspark.DataFrameReader. Moreover, spark.read makes sense because on Databricks, spark references current Spark session

(pyspark.sql.SparkSession) and spark.read therefore returns a DataFrameReader (also see documentation below). Finally, there is only one correct answer option remaining.

More info:

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

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

- pyspark.sql.SparkSession.read — PySpark 3.1.2 documentation

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