Databricks DATABRICKS-CERTIFIED-ASSOCIATE-DEVELOPER-FOR-APACHE-SPARK Online Practice Questions and Exam Preparation

Databricks DATABRICKS-CERTIFIED-ASSOCIATE-DEVELOPER-FOR-APACHE-SPARK Online Questions & Answers

• Question 1:

  Which of the following code blocks returns a DataFrame that is an inner join of DataFrame itemsDf and DataFrame transactionsDf, on columns itemId and productId, respectively and in which every itemId just appears once?

  A. itemsDf.join(transactionsDf, "itemsDf.itemId==transactionsDf.productId").distinct("itemId")
  B. itemsDf.join(transactionsDf, itemsDf.itemId==transactionsDf.productId).dropDuplicates(["itemId"])
  C. itemsDf.join(transactionsDf, itemsDf.itemId==transactionsDf.productId).dropDuplicates("itemId")
  D. itemsDf.join(transactionsDf, itemsDf.itemId==transactionsDf.productId, how="inner").distinct(["itemId"])
  E. itemsDf.join(transactionsDf, "itemsDf.itemId==transactionsDf.productId", how="inner").dropDuplicates(["itemId"])
  B. itemsDf.join(transactionsDf, itemsDf.itemId==transactionsDf.productId).dropDuplicates(["itemId"])

• Question 2:

  Which is the highest level in Spark's execution hierarchy?

  A. Task
  B. Executor
  C. Slot
  D. Job
  E. Stage
  D. Job

• Question 3:

  The code block displayed below contains an error. The code block should merge the rows of DataFrames transactionsDfMonday and transactionsDfTuesday into a new DataFrame, matching column names and inserting null values where column names do not appear in both DataFrames. Find the error.

  Sample of DataFrame transactionsDfMonday:

  1.+-------------+---------+-----+-------+---------+----+

  2.|transactionId|predError|value|storeId|productId| f|

  3.+-------------+---------+-----+-------+---------+----+

  4.| 5| null| null| null| 2|null|

  5.| 6| 3| 2| 25| 2|null|

  6.+-------------+---------+-----+-------+---------+----+

  Sample of DataFrame transactionsDfTuesday:

  1.+-------+-------------+---------+-----+

  2.|storeId|transactionId|productId|value|

  3.+-------+-------------+---------+-----+ 4.| 25| 1| 1| 4|

  5.| 2| 2| 2| 7|

  6.| 3| 4| 2| null|

  7.| null| 5| 2| null|

  8.+-------+-------------+---------+-----+

  Code block:

  sc.union([transactionsDfMonday, transactionsDfTuesday])

  A. The DataFrames' RDDs need to be passed into the sc.union method instead of the DataFrame variable names.
  B. Instead of union, the concat method should be used, making sure to not use its default arguments.
  C. Instead of the Spark context, transactionDfMonday should be called with the join method instead of the union method, making sure to use its default arguments.
  D. Instead of the Spark context, transactionDfMonday should be called with the union method.
  E. Instead of the Spark context, transactionDfMonday should be called with the unionByName method instead of the union method, making sure to not use its default arguments.
  E. Instead of the Spark context, transactionDfMonday should be called with the unionByName method instead of the union method, making sure to not use its default arguments.

  ---

  Correct code block:

  transactionsDfMonday.unionByName(transactionsDfTuesday, True) Output of correct code block:

  +-------------+---------+-----+-------+---------+----+ |transactionId|predError|value|storeId|productId| f| +-------------+---------+-----+-------+---------+----+ | 5| null| null| null| 2|null|

  | 6| 3| 2| 25| 2|null|

  | 1| null| 4| 25| 1|null|

  | 2| null| 7| 2| 2|null|

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

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

  +-------------+---------+-----+-------+---------+----+ For solving this question, you should be aware of the difference between the DataFrame.union() and DataFrame.unionByName() methods. The first one matches columns independent of their

  names, just by their order. The second one matches columns by their name (which is asked for in the question). It also has a useful optional argument, allowMissingColumns.

  This allows you to

  merge DataFrames that have different columns - just like in this example. sc stands for SparkContext and is automatically provided when executing code on Databricks. While sc.union() allows you to join RDDs, it is not the right choice for

  joining DataFrames. A hint away

  from sc.union() is given where the talks about joining "into a new DataFrame".

  concat is a method in pyspark.sql.functions. It is great for consolidating values from different columns, but has no place when trying to join rows of multiple DataFrames. Finally, the join method is a contender here. However, the default join

  defined for that method is an inner join which does not get us closer to the goal to match the two DataFrames as instructed,

  especially given that with the default arguments we cannot define a join condition.

  More info:

  - pyspark.sql.DataFrame.unionByName -- PySpark 3.1.2 documentation

  -pyspark.SparkContext.union -- PySpark 3.1.2 documentation

  -pyspark.sql.functions.concat -- PySpark 3.1.2 documentation

  Static notebook | Dynamic notebook: See test 3, 45 (Databricks import instructions)

• Question 4:

  Which of the following code blocks prints out in how many rows the expression Inc. appears in the string-type column supplier of DataFrame itemsDf?

  A. 1.counter = 0 2. 3.for index, row in itemsDf.iterrows(): 4. if 'Inc.' in row['supplier']: 5. counter = counter + 1 6. 7.print(counter)
  B. 1.counter = 0 2. 3.def count(x): 4. if 'Inc.' in x['supplier']: 5. counter = counter + 1 6. 7.itemsDf.foreach(count) 8.print(counter)
  C. print(itemsDf.foreach(lambda x: 'Inc.' in x))
  D. print(itemsDf.foreach(lambda x: 'Inc.' in x).sum())
  E. 1.accum=sc.accumulator(0) 2. 3.def check_if_inc_in_supplier(row): 4. if 'Inc.' in row['supplier']: 5. accum.add(1) 6. 7.itemsDf.foreach(check_if_inc_in_supplier) 8.print(accum.value)
  E. 1.accum=sc.accumulator(0) 2. 3.def check_if_inc_in_supplier(row): 4. if 'Inc.' in row['supplier']: 5. accum.add(1) 6. 7.itemsDf.foreach(check_if_inc_in_supplier) 8.print(accum.value)

  ---

  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 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, 22 (Databricks import instructions)

• Question 5:

  Which of the following code blocks returns a new DataFrame with only columns predError and values of every second row of DataFrame transactionsDf?

  Entire DataFrame transactionsDf:

  1.+-------------+---------+-----+-------+---------+----+

  2.|transactionId|predError|value|storeId|productId| f|

  3.+-------------+---------+-----+-------+---------+----+

  4.| 1| 3| 4| 25| 1|null|

  5.| 2| 6| 7| 2| 2|null|

  6.| 3| 3| null| 25| 3|null|

  7.| 4| null| null| 3| 2|null|

  8.| 5| null| null| null| 2|null|

  9.| 6| 3| 2| 25| 2|null|

  10.+-------------+---------+-----+-------+---------+----+

  A. transactionsDf.filter(col("transactionId").isin([3,4,6])).select([predError, value])
  B. transactionsDf.select(col("transactionId").isin([3,4,6]), "predError", "value")
  C. transactionsDf.filter("transactionId" % 2 == 0).select("predError", "value")
  D. transactionsDf.filter(col("transactionId") % 2 == 0).select("predError", "value") (Correct)
  E. 1.transactionsDf.createOrReplaceTempView("transactionsDf") 2.spark.sql("FROM transactionsDf SELECT predError, value WHERE transactionId % 2 = 2")
  F. transactionsDf.filter(col(transactionId).isin([3,4,6]))
  D. transactionsDf.filter(col("transactionId") % 2 == 0).select("predError", "value") (Correct)

• Question 6:

  Which of the following describes a valid concern about partitioning?

  A. A shuffle operation returns 200 partitions if not explicitly set.
  B. Decreasing the number of partitions reduces the overall runtime of narrow transformations if there are more executors available than partitions.
  C. No data is exchanged between executors when coalesce() is run.
  D. Short partition processing times are indicative of low skew.
  E. The coalesce() method should be used to increase the number of partitions.
  A. A shuffle operation returns 200 partitions if not explicitly set.

• Question 7:

  Which of the following code blocks reads in the JSON file stored at filePath, enforcing the schema expressed in JSON format in variable json_schema, shown in the code block below?

  Code block: 1.json_schema = """ 2.{"type": "struct",

  3.

  "fields": [

  4.

  {

  5.

  "name": "itemId",

  6.

  "type": "integer",

  7.

  "nullable": true,

  8.

  "metadata": {}

  9.

  },

  10.

  {

  11.

  "name": "supplier",

  12.

  "type": "string",

  13.

  "nullable": true,

  14.

  "metadata": {}

  15.

  }

  16.

  ] 17.}

  18."""

  A. spark.read.json(filePath, schema=json_schema)
  B. spark.read.schema(json_schema).json(filePath) 1.schema = StructType.fromJson(json.loads(json_schema)) 2.spark.read.json(filePath, schema=schema)
  C. spark.read.json(filePath, schema=schema_of_json(json_schema))
  D. spark.read.json(filePath, schema=spark.read.json(json_schema))
  C. spark.read.json(filePath, schema=schema_of_json(json_schema))

• Question 8:

  Which of the following code blocks sorts DataFrame transactionsDf both by column storeId in ascending and by column productId in descending order, in this priority?

  A. transactionsDf.sort("storeId", asc("productId"))
  B. transactionsDf.sort(col(storeId)).desc(col(productId))
  C. transactionsDf.order_by(col(storeId), desc(col(productId)))
  D. transactionsDf.sort("storeId", desc("productId"))
  E. transactionsDf.sort("storeId").sort(desc("productId"))
  D. transactionsDf.sort("storeId", desc("productId"))

  ---

  In this 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, 22 (Databricks import instructions)

• Question 9:

  Which of the following is a viable way to improve Spark's performance when dealing with large amounts of data, given that there is only a single application running on the cluster?

  A. Increase values for the properties spark.default.parallelism and spark.sql.shuffle.partitions
  B. Decrease values for the properties spark.default.parallelism and spark.sql.partitions
  C. Increase values for the properties spark.sql.parallelism and spark.sql.partitions
  D. Increase values for the properties spark.sql.parallelism and spark.sql.shuffle.partitions
  E. Increase values for the properties spark.dynamicAllocation.maxExecutors, spark.default.parallelism, and spark.sql.shuffle.partitions
  A. Increase values for the properties spark.default.parallelism and spark.sql.shuffle.partitions

• Question 10:

  Which of the following code blocks generally causes a great amount of network traffic?

  A. DataFrame.select()
  B. DataFrame.coalesce()
  C. DataFrame.collect()
  D. DataFrame.rdd.map()
  E. DataFrame.count()
  C. DataFrame.collect()