Free Databricks Certified Associate Developer for Apache Spark Sample Questions

Correct answer: A — spark.sql.shuffle.partitions

• A. Correct. spark.sql.shuffle.partitions is the specific Spark SQL/DataFrame configuration that sets the default number of partitions to use for shuffle operations caused by wide transformations like joins and aggregations. By default, this is set to 200.
• B. Incorrect. spark.default.parallelism controls the default number of partitions for RDD operations and transformations like parallelize. For Spark SQL and DataFrame operations, Spark uses spark.sql.shuffle.partitions for shuffles.
• C. Incorrect. spark.sql.files.maxPartitionBytes determines the maximum size (in bytes) of a partition when reading data from file-based sources, affecting the initial input partitioning rather than shuffle partitioning.
• D. Incorrect. spark.executor.cores configures the number of CPU cores allocated per executor, which determines how many tasks can run in parallel, but it does not define the number of shuffle partitions.
• E. Incorrect. spark.sql.adaptive.enabled enables Adaptive Query Execution (AQE). While AQE can dynamically adjust and coalesce shuffle partitions at runtime based on statistics, the property that defines the initial default number is spark.sql.shuffle.partitions.

Correct answer: B — df.rdd.getNumPartitions()

• A. Incorrect. df.countPartitions() is not a valid method in the Spark/PySpark DataFrame API. There is no built-in countPartitions() method on a DataFrame object.
• B. Correct. The standard way to determine the number of partitions for a DataFrame in PySpark is to access its underlying RDD using the .rdd attribute and then calling the getNumPartitions() method. This is an inexpensive metadata check that returns an integer representing the partition count.
• C. Incorrect. df.partitions.length is not a valid property for PySpark DataFrames. In the Scala API, one could use df.rdd.partitions.length, but df.partitions.length (without the .rdd) is invalid in both Python and Scala.
• D. Incorrect. The printSchema() method is used to display the structural metadata (column names, types, and nullability) of the DataFrame. It does not provide any information regarding the physical partitioning of the data.
• E. Incorrect. In PySpark, calling len(df) will raise a TypeError because DataFrames do not implement the __len__ method. Furthermore, even the df.count() method returns the number of rows in the DataFrame, not the number of partitions.

Correct answer: D — salesDf.persist(StorageLevel.MEMORY_ONLY)

• A. This option persists the DataFrame using the MEMORY_AND_DISK level. If memory is insufficient, Spark will spill the remaining partitions to disk. This prevents recomputation but violates the requirement that data should be dropped and recomputed when needed.
• B. In Spark 3.x, calling .cache() on a DataFrame defaults to the MEMORY_AND_DISK storage level. While this improves performance for subsequent actions, it will spill to disk if memory is full, rather than dropping the data for recomputation.
• C. This option persists the data only on disk. This does not use memory for performance and incurs significant I/O overhead, failing the primary requirement to use memory.
• D. This is the correct choice. The MEMORY_ONLY storage level stores the DataFrame partitions in memory. If a partition does not fit in the allocated memory, it is not cached and will be recomputed from its lineage (the set of transformations) every time it is needed.
• E. There is no .store() method in the Spark DataFrame API; this snippet is invalid and will cause an error.

Correct answer: B — The Executors perform a full shuffle, exchanging data across the network to create 100 new partitions.

• A. Incorrect. The Driver node is responsible for coordinating the job and managing the SparkContext; it does not process or compress data locally during a repartition. Data processing and file writing are distributed tasks performed by the Executors.
• B. Correct. In Spark, `repartition(n)` triggers a full shuffle, which is a process where data is redistributed across the network among executors. This creates exactly 'n' new partitions (in this case, 100), allowing the subsequent write operation to utilize more cores and increase parallelism.
• C. Incorrect. This describes the behavior of `coalesce`, which attempts to reduce the partition count by moving data within existing executors to minimize shuffling. Furthermore, since the developer is trying to increase the number of partitions from 2 to 100, `coalesce` would not be effective, and `repartition` must be used to trigger the necessary shuffle.
• D. Incorrect. The Cluster Manager handles resource allocation (like YARN or Kubernetes), but calling `repartition` does not change the physical size of the cluster or the number of worker nodes. It only changes how the data is partitioned within the existing allocated resources.
• E. Incorrect. Tasks are executed on the Worker nodes (Executors), not on the Driver. While the Driver manages the scheduling of the 100 resulting tasks, it does not process the partitions itself.

Correct answer: B — transactions_df.rdd

• A. Incorrect. There is no toRDD() method in the Spark DataFrame API. While the reverse conversion (RDD to DataFrame) often uses .toDF(), the conversion from DataFrame to RDD uses a property, not a method.
• B. Correct. The .rdd property is the standard way to access the underlying RDD[Row] from a DataFrame in both PySpark and Scala. This allows the developer to utilize low-level RDD operations like map, mapPartitions, or flatMap. It should be noted that converting to an RDD bypasses Catalyst optimizations and can lead to lower performance compared to DataFrame operations.
• C. Incorrect. sparkCore() is not a valid method in the DataFrame API. To interact with Spark Core features from a SparkSession, you would typically use spark.sparkContext.
• D. Incorrect. asRDD is not a valid property or method. While Scala DataSets have an as[T] method for typed conversions, there is no corresponding asRDD property; the correct accessor is .rdd.
• E. Incorrect. The SparkSession object does not have a toRDD method that accepts a DataFrame as an argument. Conversion is handled via the DataFrame's own properties.

Correct answer: C — Lazy evaluation

• A. Incorrect. In-memory computation refers to Spark's ability to cache datasets in RAM for faster iterative processing. This does not explain why the code executes instantly without jobs, as materializing data in memory still requires job execution via an action.
• B. Incorrect. Fault tolerance is Spark's ability to recover lost data partitions using lineage and recomputation. This feature handles resilience during execution but does not explain why transformations are deferred.
• C. Correct. Lazy evaluation means that Spark does not execute transformations immediately. Instead, it builds a logical execution plan (DAG) and waits until an action (like count, collect, or save) is called to trigger the actual computation. Since the provided code only contains transformations (read, filter, and select), no Spark jobs are launched in the UI.
• D. Incorrect. Data skew refers to an uneven distribution of data across partitions, which can cause performance bottlenecks during active job execution. It does not prevent jobs from being created.
• E. Incorrect. Predicate pushdown is an optimization technique where filtering is pushed to the data source (like Parquet) to reduce I/O. While this occurs in the background for this code, it is part of the execution plan that is only triggered when an action is invoked.

Correct answer: C — No, it only prints the metadata (schema) stored in the DataFrame object on the driver.

• A. Incorrect. `df.printSchema()` does not read all the data to determine the schema. The schema is stored in the DataFrame's metadata on the driver. Even if schema inference occurred during the initial data load (e.g., reading a CSV with `inferSchema` set to true), that Spark job would have been triggered at the time of the read operation, not by the call to `printSchema()` itself.
• B. Incorrect. Reading the first 20 rows is the default behavior of the `df.show()` action, which collects a sample of data to the driver. `printSchema()` does not execute a job to fetch rows from the executors.
• C. Correct. `printSchema()` is a utility method that prints the DataFrame's schema metadata already held on the driver. It does not launch a Spark job or perform any distributed computation on the data.
• D. Incorrect. While it is true that `printSchema()` does not trigger a Spark job, it is not a transformation. Transformations (like `map` or `filter`) are lazy and return a new DataFrame. `printSchema()` is a helper method that returns `None` (or `Unit` in Scala) and prints directly to the console; it is not part of the lazy execution plan.
• E. Incorrect. Shuffles are triggered by wide transformations (e.g., `groupBy`, `join`, `repartition`) that require data to be redistributed across the cluster. `printSchema()` only accesses metadata on the driver and involves no data movement.

Correct answer: E — Line 5

• A. Incorrect. The `spark.read.load()` method is a transformation that defines a data source. Due to Spark's lazy evaluation, this line only adds an operation to the logical plan (DAG) and does not trigger a job to load the data. The actual data processing is deferred until an action is invoked.
• B. Incorrect. The `filter()` method is a narrow transformation. Like other transformations in Spark, it is evaluated lazily. This line simply adds another step to the execution plan without triggering a job.
• C. Incorrect. The `withColumnRenamed()` method is a narrow transformation that modifies the DataFrame's schema. It is a lazy operation that updates the logical plan but does not cause a Spark Job to be executed.
• D. Incorrect. The `orderBy()` method is a wide transformation, as it requires a shuffle to sort the data across partitions. However, it is still a transformation and is subject to lazy evaluation. The job execution is deferred until an action is called.
• E. Correct. The `show()` method is an action, which forces Spark to execute the entire chain of lazy transformations defined in the preceding lines. According to the official Databricks Spark Developer exam guide, understanding lazy evaluation and actions is a key concept. Actions are the operations that trigger the computation and creation of a Spark Job.

Correct answer: B — products_df.createOrReplaceTempView("products")

• A. Incorrect. This method creates a global temporary view, which is registered in the `global_temp` database and is available across all SparkSessions in the cluster. It also does not handle the replacement of an existing view; it would throw an AnalysisException if the name is already taken.
• B. Correct. The `createOrReplaceTempView` method creates a local temporary view scoped to the current SparkSession and ensures that any existing view with the same name is replaced.
• C. Incorrect. This is not a valid method in the Spark API. The SparkSession object (`spark`) does not have a `createView` method; views are registered using the DataFrame API.
• D. Incorrect. While `createTempView` is scoped to the current SparkSession, it will throw an AnalysisException if a view with the same name already exists. It does not satisfy the requirement to replace an existing view.
• E. Incorrect. This method writes the DataFrame to a persistent table in the metastore. Persistent tables are available across sessions and clusters and remain after the SparkSession is terminated, which violates the requirement for it to be session-scoped only.

Correct answer: B — +--------+-----------+ |category|total_value| +--------+-----------+ | A| 300| | B| 150| +--------+-----------+

• A. Incorrect. While the aggregated sums (300 for A and 150 for B) are correct, this option shows the rows sorted with 'B' before 'A'. The query specifies `ORDER BY category`, which defaults to ascending order (A then B).
• B. Correct. The query groups the data by `category`. Category 'A' has two records with values 100 and 200, resulting in a sum of 300. Category 'B' has one record with a value of 150. The `ORDER BY category` clause ensures that 'A' appears before 'B' in the output.
• C. Incorrect. The query explicitly provides an alias using `AS total_value`. Furthermore, Spark SQL does not require aliases for aggregate functions; if omitted, Spark would generate a default column name like `sum(value)`.
• D. Incorrect. This option shows the raw, unaggregated data. Because the query uses `GROUP BY category`, the output must collapse the original rows into one row per unique category.
• E. Incorrect. Spark SQL temporary views behave like standard relational tables and fully support all standard SQL aggregate functions, including `SUM`.

Correct answer: A — The DataFrame will have a single column named `value` of type String.

• A. Correct. When using `spark.read.text()`, Spark returns a DataFrame with a single column named `value` of type String. Each row in the DataFrame corresponds to exactly one line from the input text files. This is the fixed, default schema for the text data source.
• B. Incorrect. While the data represents lines, Spark uses the standardized column name `value` by default, not `line`.
• C. Incorrect. The text datasource does not attempt to parse fields or infer columns based on delimiters; it treats every character in a line as part of a single string. Delimiter-based inference is a feature of the CSV datasource.
• D. Incorrect. There are no default `text` or `id` columns generated. Users would need to manually add an `id` using functions like `monotonically_increasing_id()` or split the `value` column after the initial read.
• E. Incorrect. The text reader does not require an explicit schema. It defaults to a single `StringType` column named `value`. Only formats like JSON or Parquet often benefit from schema inference or explicit definitions, whereas text is inherently simple.

Correct answer: C — /mnt/sales/year=2023/month=01/part-0000.parquet

• A. Incorrect. Spark's partitionBy method does not concatenate multiple partition columns into a single directory name with underscores. It creates a nested hierarchical structure.
• B. Incorrect. While this shows nested folders, Spark's standard partitioning (Hive-style) includes the column name in the folder name (e.g., year=2023), not just the raw value. This allows Spark to automatically infer the column names and types when reading the data back.
• C. Correct. Spark uses Hive-style partitioning by default. When using partitionBy, it creates a directory structure where each level is named using the 'columnName=value' convention. For the columns 'year' and 'month', this results in /mnt/sales/year=2023/month=01/ followed by the data files.
• D. Incorrect. Spark uses equals signs (=) to separate column names from values in directory paths, not hyphens (-).
• E. Incorrect. Spark does not add a 'partition_' prefix to directory names. It uses the exact column names provided in the partitionBy method.

Correct answer: A — df.write.mode("append").parquet("/mnt/history")

• A. Correct. This is the standard, cross-language Spark DataFrameWriter API pattern. It uses .mode("append") to set the write behavior and .parquet(path) as the terminal method to write the data to the specified location.
• B. Incorrect. In Spark's DataFrameWriter, the save mode is an internal state set specifically by the .mode() method. It cannot be set using the generic .option() method, which is intended for data source-specific configurations.
• C. Incorrect. The DataFrameWriter class does not have an .append() method. 'append' is a value passed to the .mode() method to configure the writer's behavior.
• D. Incorrect. Although PySpark's implementation of the parquet() method optionally accepts a mode parameter, the Databricks Spark Developer Associate exam focuses on the standard, idiomatic fluent API pattern: df.write.mode(...).parquet().
• E. Incorrect. The save() method is a member of the DataFrameWriter class, which is accessed via the .write property of a DataFrame (i.e., df.write.save()). It is not a method available directly on the DataFrame object.

Correct answer: A — current_date()

• A. Correct. The current_date() function returns the current date at the start of query evaluation as a DateType column (year-month-day), specifically without time components.
• B. Incorrect. The current_timestamp() function returns the current timestamp including both date and time (hours, minutes, seconds, and fractional seconds), which does not meet the requirement of being date-only.
• C. Incorrect. The now() function is an alias for current_timestamp() and returns a TimestampType with time information included.
• D. Incorrect. There is no built-in today() function in the Spark SQL or DataFrame API. Attempting to use it would result in an analysis error.
• E. Incorrect. date_now() is not a recognized built-in function in Spark. The standard function for retrieving the execution date is current_date().

Correct answer: B — spark.read.option("mode", "PERMISSIVE").option("columnNameOfCorruptRecord", "_bad_record").json(path)

• A. Incorrect. The `FAILFAST` mode causes the read job to abort and throw an exception immediately upon encountering any malformed record. This fails the requirement to keep the job running.
• B. Correct. `PERMISSIVE` is the default mode which keeps the job running and sets malformed fields to null. The option `columnNameOfCorruptRecord` allows the developer to specify a column name (in this case, `_bad_record`) to store the original raw JSON string for records that could not be parsed. Note that for this column to be populated, the provided schema must also include this column name.
• C. Incorrect. `DROPMALFORMED` drops records that contain errors entirely rather than setting fields to null. Additionally, `badRecordsPath` is used to specify a directory on the filesystem to write corrupt records to, rather than creating a column within the DataFrame.
• D. Incorrect. While `PERMISSIVE` mode is correct, `badRecords` is not a valid option name for the JSON reader to capture raw data into a specific column; the correct option is `columnNameOfCorruptRecord`.
• E. Incorrect. `IGNORE` is not a valid JSON parsing mode in Spark (the standard modes are PERMISSIVE, DROPMALFORMED, and FAILFAST), and `recordLog` is not a recognized option for capturing corrupt records.

Correct answer: C — `df_users` will remain in its existing partitions, and `df_roles` will be replicated to every executor hosting a partition of `df_users`.

• A. Incorrect. This statement describes a Shuffle Hash Join or Sort-Merge Join. The primary advantage of a broadcast join is that it avoids a network-intensive shuffle of the large dataset (df_users).
• B. Incorrect. In a Broadcast Hash Join, df_users is not shuffled; it is processed in its current partitions. Additionally, while df_roles is initially collected to the driver, it is then broadcast (replicated) to all executors, not kept solely on the driver.
• C. Correct. In a Broadcast Hash Join (also known as a map-side join), the small DataFrame (df_roles) is collected to the driver and then replicated to every executor. This allows Spark to join each partition of the large DataFrame (df_users) locally without moving its data across the network.
• D. Incorrect. Spark does not require the large side of the join to fit in memory or be broadcast. Only the side being broadcasted (df_roles, which is only 1KB) must fit within the broadcast memory limits.
• E. Incorrect. A Cross Join is used when there is no join condition (Cartesian product). In this snippet, a join condition on "role_id" is explicitly provided, and the size disparity makes it a perfect candidate for a Broadcast Hash Join, not a Cross Join.

Correct answer: C — Spark performs the union by position, resulting in data corruption (e.g., names in the id column) for the rows from `df2`.

• A. Incorrect. Spark does not automatically reorder columns based on names when using the `union` method. To achieve automatic alignment by column name, the `unionByName` method must be used instead.
• B. Incorrect. Spark's `union` requires the same number of columns and compatible data types. If these conditions are met, it will not throw an AnalysisException just because the column names are in a different order; it will simply process them by position.
• C. Correct. The `union` operation in Spark is resolved by position (column index). Since `df1` expects `id` in the first position and `name` in the second, but `df2` provides them in reverse order, the data from `df2` will be inserted into the wrong columns, leading to data corruption/misalignment.
• D. Incorrect. A union is a set operation (vertical concatenation), whereas a full outer join is a relational operation based on a join key. Spark will not change the requested union into a join.
• E. Incorrect. Spark's `union` does not drop columns based on their position or name. It appends the rows as they are ordered in the respective DataFrames.

Correct answer: B — rawDf.withColumn("is_processed", lit(True))

• A. Incorrect. The `withColumn` method expects a Column object as its second argument. Passing a native Python boolean `True` directly will result in a TypeError because it is not a Spark Column expression.
• B. Correct. The `lit()` function (short for literal) is used to create a Column expression from a constant value. Using `lit(True)` correctly wraps the boolean value so it can be used within the `withColumn` method.
• C. Incorrect. The `col()` function is used to reference an existing column by its name (string). It cannot be used to transform a literal boolean into a Column object.
• D. Incorrect. In Python, the boolean `True` is a primitive type and does not possess an `.alias()` method. To use this approach, one would need to use `lit(True).alias("is_processed")`.
• E. Incorrect. The DataFrame API does not have a method named `addColumn`. The standard method to add or replace a column is `withColumn`.

Correct answer: A — groupby().applyInPandas()

• A. Correct. The groupby().applyInPandas() method (also known as Grouped Map Pandas UDF) is designed for this exact use case. It allows you to split the data into groups, apply a Python function that takes a pandas.DataFrame and returns a pandas.DataFrame for each group, and define a new output schema that can differ from the input schema.
• B. Incorrect. While mapInPandas() exists, it is a method of the DataFrame class, not the GroupedData class. It is used to apply a transformation to an iterator of pandas.DataFrames representing partitions, rather than logical groups defined by a key.
• C. Incorrect. The groupby().agg() method is intended for aggregation operations that reduce each group to a single row (e.g., sum, mean, or Grouped Aggregate Pandas UDFs). It cannot be used to return a full transformed DataFrame per group with an arbitrary schema.
• D. Incorrect. There is no select().apply() method in the PySpark DataFrame API. Selecting columns and applying transformations is typically done through withColumn or select using functions and UDFs, but not in a grouped-map context.
• E. Incorrect. Combining withColumn() with a pandas_udf is used for vectorized scalar or series operations (mapping input columns to an output column). It does not support the group-to-DataFrame transformation pattern where the output schema can be completely different from the input.

Correct answer: C — It uses the HyperLogLog++ algorithm to provide an estimated cardinality.

• A. Incorrect. As the name implies, `approx_count_distinct` returns an estimated count of distinct items rather than an exact result. For exact results, `count_distinct` (or `count(distinct ...)`) should be used, though it is more computationally expensive.
• B. Incorrect. `approx_count_distinct` is typically much faster and more memory-efficient than `countDistinct` because it does not require a full global shuffle of all unique values. However, it provides lower precision because it is an estimation.
• C. Correct. Spark's `approx_count_distinct` function utilizes the HyperLogLog++ (HLL++) algorithm to estimate the number of unique elements in a large dataset with a small, tunable margin of error (controlled by the rsd parameter).
• D. Incorrect. Like most aggregate functions in Apache Spark, `approx_count_distinct` ignores null values during its calculation and does not throw an error.
• E. Incorrect. The function is not limited to integer columns; it can be applied to various data types, including strings, as it internally hashes the values to compute the cardinality estimate.

Correct answer: A — transactionsDf.sort(col("storeId").asc(), col("amount").desc())

• A. Correct. This code snippet uses the `sort` method with explicit Column references (`col()`). It correctly specifies ascending order for `storeId` and descending order for `amount` using the `asc` and `desc` methods. This syntax is consistent and works in both Scala and Python APIs.
• B. Incorrect. This code snippet contains a syntax error because the `.desc` method cannot be called on a string literal like `"amount"`. It must be called on a Column object (e.g., `col("amount").desc()`).
• C. Incorrect. This code snippet mixes a string argument (`"storeId"`) with a Column object produced by the `desc()` function. While PySpark's flexible API might allow this, the standard Spark (Scala) Dataset API does not have a signature that allows mixing strings and Column objects in the same varargs list; it requires either all strings or all Columns.
• D. Incorrect. The DataFrame API does not have a method named `order`. The correct methods for sorting are `sort` or `orderBy`.
• E. Incorrect. The `sortBy` method is used for RDDs or specifically typed Datasets in Scala (often requiring a lambda function), but it is not the standard method for untyped DataFrame sorting across columns.

Correct answer: B — It will contain 3 rows. The row with item_id 101 will appear once, but it is non-deterministic whether the color will be Red or Green.

• A. Incorrect. This would be the result if no columns were passed to dropDuplicates() or if dropDuplicates() was called without arguments on a DataFrame where all rows are unique. However, because a subset ["item_id"] is provided, rows with the same item_id (101) are considered duplicates despite having different colors.
• B. Correct. The dropDuplicates(["item_id"]) method identifies duplicates based only on the specified columns. Since item_id 101 appears twice, one instance will be dropped, resulting in 3 total rows (101, 102, 103). Because Spark DataFrames are distributed and unordered, and dropDuplicates does not guarantee which row is kept, the value in the 'color' column for item_id 101 is non-deterministic.
• C. Incorrect. Spark does not guarantee the retention of the first-appearing row. DataFrames are distributed across partitions, and without an explicit sort and a deterministic operation (like a Window function), the row selected for retention is not guaranteed.
• D. Incorrect. The dropDuplicates operation is designed to keep one representative row for every set of duplicates, not to filter out all rows containing duplicated keys.
• E. Incorrect. The dropDuplicates method (and its alias drop_duplicates) is explicitly designed to accept an optional subset of columns as a list of strings. It does not require all columns to be specified.

Correct answer: B — logsDf.filter("timestamp IS NOT NULL AND level != 'DEBUG'")

• A. Incorrect. This snippet uses the logical OR operator (|). The requirement specifies that both conditions must be met (timestamp is not null AND level is not 'DEBUG'), which requires a logical AND (&).
• B. Correct. Spark's filter method accepts SQL-style string expressions. The expression 'timestamp IS NOT NULL AND level != 'DEBUG'' correctly applies both validation requirements using standard SQL operators.
• C. Incorrect. Although dropna(subset=["timestamp"]) effectively removes null timestamps, the subsequent filter(col("level") == "DEBUG") retains only the 'DEBUG' rows, which is the exact opposite of the requirement to exclude them.
• D. Incorrect. The select method is used for column projection. Instead of filtering rows, this code would return a DataFrame with two boolean columns indicating whether the conditions were met for each row.
• E. Incorrect. The method isValid() is not a member of the Spark Column class; the standard method for null checking is isNotNull().

Correct answer: C — The value will be `30` because the `map` transformation is executed for each action (`collect()` and `count()`) due to Spark's lazy evaluation.

• A. Incorrect. Accumulators are specifically designed to be safely updated inside transformations like `map` or `foreach` that run on distributed executors. Their primary purpose is to aggregate values from worker nodes back to the driver.
• B. Incorrect. While Spark has many optimizations, it will re-evaluate the entire transformation lineage for each action by default. To prevent this re-computation, the RDD or DataFrame must be explicitly persisted in memory or on disk using `.cache()` or `.persist()`.
• C. Correct. Spark uses lazy evaluation, meaning transformations are only computed when an action is called. Since `processed_data` is not cached, the `map` transformation is executed once for the `collect()` action (adding 15 to the accumulator) and a second time for the `count()` action (adding another 15), resulting in a final value of 30. The official documentation recommends using `.cache()` to avoid this redundant computation.
• D. Incorrect. The rule is that accumulators are write-only *within tasks running on executors*. Tasks can add to an accumulator but cannot read its value. The final, aggregated value can only be read on the driver program after the actions have completed, which is what `print(acc.value)` correctly does.
• E. Incorrect. Both `collect()` and `count()` are actions that trigger the execution of the full Directed Acyclic Graph (DAG) of transformations. A `count()` on an RDD that results from a `map` requires executing the `map` to determine how many elements are in the final RDD, thus triggering the accumulator logic again.

Correct answer: C — df = df.repartition(col("user_id"))

• A. Incorrect. While `sort()` (an alias for `orderBy`) triggers a shuffle and uses range partitioning, it is computationally more expensive and less efficient than hash-partitioning for preparing data for subsequent joins and aggregations on the same key.
• B. Incorrect. `coalesce(1)` reduces the number of partitions to one, which would place all data in a single partition. This destroys parallelism and creates a severe performance bottleneck without providing the benefits of key-based partitioning.
• C. Correct. `repartition(col("user_id"))` performs a hash-based shuffle to partition the DataFrame by the specific column. This ensures that all rows with the same `user_id` are co-located on the same partition, significantly reducing shuffle overhead for subsequent joins and aggregations on that column.
• D. Incorrect. `cache()` persists the DataFrame in memory or disk in its current state. It does not change how the data is partitioned across the cluster.
• E. Incorrect. `orderBy()` triggers a global sort using range partitioning. This is a heavy operation intended for sorting output rather than optimizing the data layout for keyed joins, where hash partitioning is the standard approach.

Correct answer: A — The executors are spending a significant amount of time performing Garbage Collection, indicating high memory pressure.

• A. Correct. In the Spark UI, the GC Time column turns red when Garbage Collection takes up more than 10-20% of the total task time. This indicates that the JVM is spending a disproportionate amount of time reclaiming memory, which is a clear sign of high memory pressure. This usually suggests the need for memory tuning, such as increasing executor memory, adjusting GC settings, or reducing object churn.
• B. Incorrect. Idle executors waiting for tasks would show low CPU usage and minimal GC activity. High GC Time implies active memory management effort by the JVM, which is the opposite of being idle.
• C. Incorrect. Slow network connections manifest as high shuffle read/write times or task communication delays. Network latency does not directly impact the JVM's internal garbage collection metrics.
• D. Incorrect. Disk I/O bottlenecks are typically indicated by high spill metrics or long task durations in I/O intensive stages. While related to performance, these are tracked separately from GC time, which specifically reflects memory reclamation pauses.
• E. Incorrect. While data skew can lead to memory pressure on specific executors (because one executor is handling much more data than others), the GC Time metric itself specifically measures memory management overhead. Data skew is better identified by comparing the distribution of task durations and shuffle sizes across all tasks in a stage.

Correct answers: A, B — spark.sql.adaptive.skewJoin.skewedPartitionFactor; spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes

• A. Correct. spark.sql.adaptive.skewJoin.skewedPartitionFactor is a multiplier used by AQE to determine skewness. A partition is considered skewed if its size is larger than this factor multiplied by the median partition size (default is 5).
• B. Correct. spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes sets the absolute size threshold. A partition must be larger than this value (default 256MB) and also meet the factor requirements to be flagged as skewed.
• C. Incorrect. spark.sql.shuffle.partitions controls the default number of shuffle partitions for operations like joins and aggregations, but it does not define the logic for skew detection.
• D. Incorrect. spark.sql.adaptive.advisoryPartitionSizeInBytes is the target size for partitions when AQE coalesces or splits them. While it influences the resulting partition size after optimization, it is not one of the thresholds used to initially detect a skewed partition.
• E. Incorrect. spark.sql.autoBroadcastJoinThreshold determines the maximum table size for a broadcast join and is unrelated to identifying skewed partitions during a shuffle-sort-merge join.

Correct answer: B — sensor_stream.writeStream.format("memory").queryName("sensor_view").start()

• A. Incorrect. While createOrReplaceTempView can be called on a streaming DataFrame, it simply registers the DataFrame's logic as a view for use in further SQL queries; it does not start an active streaming query or materialize data into a table for interactive inspection in separate cells.
• B. Correct. To query a stream using Spark SQL, you use the 'memory' sink. Specifying .format("memory") identifies the sink, and .queryName("sensor_view") defines the name of the in-memory table that will be registered in the Spark catalog for SQL querying.
• C. Incorrect. The console sink prints output directly to the driver's standard output (stdout) and does not store the data in an in-memory table for SQL querying. Furthermore, 'tableName' is not a valid option for the console sink.
• D. Incorrect. saveAsTable is a method of the batch DataFrameWriter (DataFrame.write), not the DataStreamWriter (DataFrame.writeStream). Additionally, the Delta format writes data to persistent storage, not memory.
• E. Incorrect. 'memory' is a sink format, not an output mode. Output modes (append, complete, update) define how data is written to the sink, while 'memory' defines where it is written. Also, the start() method for a memory sink requires the table name to be set via queryName().

Correct answer: C — .trigger(processingTime="30 seconds")

• A. Incorrect. The `.trigger(once=True)` (or the newer `availableNow=True`) option configures the query to process all available data in a single micro-batch and then stop. It does not set up a recurring schedule.
• B. Incorrect. The `.trigger(continuous="30 seconds")` option refers to continuous processing mode, which is an experimental low-latency processing engine (sub-millisecond) that works differently than the standard micro-batch model.
• C. Correct. The `.trigger(processingTime="30 seconds")` option sets a processing-time micro-batch trigger. This schedules micro-batches at the specified fixed interval (e.g., every 30 seconds). If a batch takes longer than the interval to complete, the next batch starts as soon as the previous one finishes.
• D. Incorrect. This is syntactically invalid. The `trigger` method expects specific named arguments like `processingTime`, `once`, or `availableNow`. It does not accept an `interval` argument.
• E. Incorrect. Configuration for the trigger mechanism must be defined using the `.trigger(...)` method. There is no standard `triggerInterval` key used within the `.option()` method for this purpose.

Correct answer: C — OrderBy (Sort)

• A. Incorrect. Filter is a stateless operation that can be applied to each incoming row independently. It is supported on streaming DataFrames as it does not require a global view of the data or persistent state across records.
• B. Incorrect. Select (projection) is a stateless operation supported on streaming DataFrames. It allows for the selection or computation of columns on a per-row basis as records arrive in the stream.
• C. Correct. OrderBy (global sort) is not supported on an unbounded streaming DataFrame because sorting requires knowledge of the entire dataset to determine rank. Since streams are continuous and unbounded, global ordering is impossible unless the data is first aggregated into a finite state (using Complete output mode) or handled in a bounded micro-batch context.
• D. Incorrect. WithColumn is a stateless transformation used to add or modify columns. It operates on each record individually without requiring aggregation or global state, making it fully supported in Structured Streaming.
• E. Incorrect. Drop is a schema-level transformation that removes columns. Like Select, it is a stateless operation applied per-row and is supported on streaming DataFrames.

Correct answer: C — Streaming `dropDuplicates` requires the state to be maintained, potentially indefinitely without watermarks; Batch `dropDuplicates` processes all data at once.

• A. Incorrect. Batch `dropDuplicates()` does **not** require a watermark; watermarks are a feature of Structured Streaming used to bound state in stateful operations. In streaming, `dropDuplicates()` can optionally use a watermark to manage state, but it is not required in batch processing, which handles data all at once.
• B. Incorrect. Streaming `dropDuplicates()` is **stateful**: it must maintain a state across triggers to identify duplicates in a continuous stream. Batch `dropDuplicates()` is **not** stateful—it processes the entire static DataFrame in one go without retaining any state between operations.
• C. Correct. According to official documentation, in Structured Streaming, `dropDuplicates()` maintains all data as intermediate state to enable global deduplication, and without watermarks this state can grow indefinitely, risking out-of-memory errors. For a batch DataFrame, `dropDuplicates()` simply removes duplicate rows from the static dataset in a single pass.
• D. Incorrect. The two functions behave very differently: batch `dropDuplicates()` is a one-time, stateless operation on a finite dataset, while streaming `dropDuplicates()` is a continuous, stateful operation that requires watermarking to manage state and handle late data effectively.
• E. Incorrect. Streaming `dropDuplicates()` supports deduplication on a **subset of multiple columns**, just like batch `dropDuplicates()`. You can pass a list of column names to the method in both batch and streaming contexts.

Correct answer: E — It runs the entire Spark application, including the driver and executors, within a single JVM process.

• A. Incorrect. Local mode does not require a remote cluster manager such as YARN, Mesos, or Kubernetes. It is designed to run on a single machine using local resources without the need for external cluster management.
• B. Incorrect. In standard local mode, the driver and executors run within the same JVM process as threads, not in separate JVMs. Running multiple JVMs on a single machine is more characteristic of a 'local-cluster' setup.
• C. Incorrect. Local mode is intended for local development, unit testing, and debugging. It is not recommended for large-scale production workloads, which require distributed cluster deployment for scalability and fault tolerance.
• D. Incorrect. While local mode avoids network communication between separate machines, it may still involve internal I/O or loopback communication during shuffles. This statement is not the defining characteristic compared to the single-process execution model.
• E. Correct. The defining characteristic of local mode is that the entire Spark application—including the driver and all executors—runs within a single Java Virtual Machine (JVM) process on a single machine.

Correct answer: C — On the Spark executor nodes.

• A. In the Spark Connect architecture, the client machine is only responsible for building the logical plan and submitting it to the Spark Connect server. The actual row-level processing of the data, such as executing a UDF, does not happen on the client.
• B. The Spark driver (or the Spark Connect server side that acts as the driver) coordinates the job and schedules tasks. However, it does not typically execute the per-row logic of a UDF on the dataset; that work is distributed.
• C. Correct. When using Spark Connect, Python UDFs are serialized and shipped to the Spark executor nodes. The executors then execute the function within Python worker processes to process data partitions in parallel across the cluster.
• D. The Spark Connect server proxy handles RPC calls and translates client requests into execution plans, but it does not execute the per-row Python UDF logic itself. It forwards the work to the cluster where executors handle the execution.
• E. Python UDFs are not translated into Scala or JVM bytecode. They are executed in their native Python runtime on the executor nodes, often involving a Python worker process and potentially using Arrow for efficient data transfer.

Correct answer: A — psdf = spark_df.to_pandas_on_spark() psdf['category'].value_counts()

• A. Correct. The `to_pandas_on_spark()` method converts a Spark DataFrame into a Pandas-on-Spark DataFrame. Operations performed on this object, such as `value_counts()`, are executed in a distributed manner across the Spark cluster, making it efficient for large (100GB) datasets.
• B. Incorrect. The `toPandas()` method collects the entire dataset into the memory of the Spark driver node to create a standard pandas DataFrame. With a 100GB dataset, this would lead to an OutOfMemory (OOM) error on the driver.
• C. Incorrect. While `pandas_api()` was introduced in Spark 3.2 as a valid method, `to_pandas_on_spark()` is the canonical method emphasized in the Databricks certification curriculum and standard documentation for the Pandas API on Spark. In the context of this exam, A is the recognized standard answer.
• D. Incorrect. `to_koalas()` is part of the legacy Koalas library which has been integrated into PySpark as the Pandas API on Spark. Furthermore, `count_values()` is not a valid method name; the correct pandas-compatible method is `value_counts()`.
• E. Incorrect. `as_pandas()` is not a valid PySpark DataFrame method. Additionally, while `groupby().count()` is a valid logic, it is not the most direct equivalent to the requested `value_counts()` operation and the conversion method used here is invalid.

Correct answer: B — A GROUPED_MAP Pandas UDF, because it allows operating on the entire data for a group (product category) at once within the function, making it ideal for stateful operations like model training.

• A. Incorrect. Scalar Pandas UDFs (Series to Series) are vectorized element-wise operations. They are not designed to receive an entire group's DataFrame for model training and cannot maintain or isolate group-specific state during execution.
• B. Correct. A GROUPED_MAP Pandas UDF (invoked via groupby().applyInPandas()) receives each group as a pandas.DataFrame. This allows the developer to fit a model per group and produce row-level predictions. This pattern naturally isolates model state to the group scope and is the standard approach for group-specific operations like training and predicting.
• C. Incorrect. SCALAR_ITER Pandas UDFs process data as an iterator of Series batches. While they are useful for optimizing UDFs that require expensive initialization (like loading a pre-trained model once per task), they are not designed to group data by key for training purposes.
• D. Incorrect. Standard Python UDFs are row-at-a-time and lack the vectorized performance of Pandas UDFs. Additionally, the statement that machine learning models cannot be serialized within a Pandas UDF is false; they are frequently used within these functions via Spark's serialization mechanisms.
• E. Incorrect. GROUPED_AGG Pandas UDFs are intended to compute a single aggregated value (a scalar) per group. They are not suitable for fitting a model and returning row-wise predictions for a full dataset.