PySpark Certification Practice Test 02
DataFrame Operations and Transformations
Question 1 of 20
📋 Multiple ChoiceQuestion 21
A newly joined Data Engineer at an e-commerce company receives a customer dataset with 45 columns. Downstream reporting requires extracting only three specific columns: `customer_id`, `first_name`, and `email`.
Which PySpark DataFrame transformation should be used to project this subset of columns?
A. filter()
B. groupBy()
C. dropDuplicates()
✓
D. select()
Answer
✅ D. select()
Explanation
1. Concept Overview In Apache Spark, column projection is handled by the **select()** transformation. It creates a new DataFrame containing only the specified columns or expressions, pruning unused columns from downstream execution.
2. Why the Correct Option is Right **select()** is specifically engineered for column projection: `df.select("customer_id", "first_name", "email")` projects the required three columns. Spark's Catalyst Optimizer pushes this projection down into the underlying file reader (e.g., Parquet or Delta), scanning only the required physical columns from storage.
3. Why Other Options Are Incorrect - **filter():** Filters rows based on a boolean condition (row-level selection); it does not project or eliminate columns. - **groupBy():** Aggregates rows based on grouping keys; it does not perform simple column projection. - **dropDuplicates():** Eliminates duplicate rows from the DataFrame; it preserves the full set of columns.
4. Production Scenario / Code Example
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("ProjectionExample").getOrCreate()
# Load customer dataset with 45 columns
df_customers = spark.read.parquet("s3://datalake/customers/")
# Project only the required columns:
df_subset = df_customers.select(
col("customer_id"),
col("first_name"),
col("email")
)
# Under Parquet/Delta, only the 3 column chunks are read from disk!5. Real-World Architecture Context Columnar storage formats (Parquet, ORC, Delta Lake) organize data by columns rather than rows. Calling `select()` early in an ETL pipeline enables **column projection pushdown**, reducing disk I/O, network transfer, and executor memory footprint by over 90% compared to scanning all 45 columns.
6. Common Pitfalls & Anti-Patterns - **Selecting columns after an expensive shuffle:** Always project columns *before* wide transformations (joins, aggregations) to avoid shuffling unneeded columns across the network. - **Passing strings with typos:** Passing non-existent column names raises an `AnalysisException`. Using `col("name")` provides cleaner syntax and supports column expressions.
7. Performance & Cost Considerations Column pruning is one of the most effective optimizations in distributed computing. Scanning 3 columns out of 45 cuts storage read requests and cluster memory consumption dramatically.
8. Exam / Interview Tip Remember: `select()` operates on **columns** (projection), while `filter()` / `where()` operates on **rows** (selection).
9. Key Takeaways - `select()` projects a subset of DataFrame columns. - Enables Catalyst column projection pushdown in columnar formats. - Always prune unneeded columns early in your pipelines.
10. Related Tutorials & Resources - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api) - [PySpark DataFrame Basics (Part 1)](/tutorials/pyspark/pyspark-dataframe-basics)`
Answered: 0 / 20
Question Navigator (Test 02)0% Answered
PySpark Certification Practice Test 02
DataFrame Operations and Transformations
Insightful Saga — Modern Data Engineering Certification Preparation
Question 21
## Question 21 A newly joined Data Engineer at an e-commerce company receives a customer dataset with 45 columns. Downstream reporting requires extracting only three specific columns: `customer_id`, `first_name`, and `email`. Which PySpark DataFrame transformation should be used to project this subset of columns?
A. filter()
B. groupBy()
C. dropDuplicates()
D. select()
Answer: ✅ D. select()
Explanation:
1. Concept Overview In Apache Spark, column projection is handled by the **select()** transformation. It creates a new DataFrame containing only the specified columns or expressions, pruning unused columns from downstream execution.
2. Why the Correct Option is Right **select()** is specifically engineered for column projection: `df.select("customer_id", "first_name", "email")` projects the required three columns. Spark's Catalyst Optimizer pushes this projection down into the underlying file reader (e.g., Parquet or Delta), scanning only the required physical columns from storage.
3. Why Other Options Are Incorrect - **filter():** Filters rows based on a boolean condition (row-level selection); it does not project or eliminate columns. - **groupBy():** Aggregates rows based on grouping keys; it does not perform simple column projection. - **dropDuplicates():** Eliminates duplicate rows from the DataFrame; it preserves the full set of columns.
4. Production Scenario / Code Example
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("ProjectionExample").getOrCreate()
# Load customer dataset with 45 columns
df_customers = spark.read.parquet("s3://datalake/customers/")
# Project only the required columns:
df_subset = df_customers.select(
col("customer_id"),
col("first_name"),
col("email")
)
# Under Parquet/Delta, only the 3 column chunks are read from disk!5. Real-World Architecture Context Columnar storage formats (Parquet, ORC, Delta Lake) organize data by columns rather than rows. Calling `select()` early in an ETL pipeline enables **column projection pushdown**, reducing disk I/O, network transfer, and executor memory footprint by over 90% compared to scanning all 45 columns.
6. Common Pitfalls & Anti-Patterns - **Selecting columns after an expensive shuffle:** Always project columns *before* wide transformations (joins, aggregations) to avoid shuffling unneeded columns across the network. - **Passing strings with typos:** Passing non-existent column names raises an `AnalysisException`. Using `col("name")` provides cleaner syntax and supports column expressions.
7. Performance & Cost Considerations Column pruning is one of the most effective optimizations in distributed computing. Scanning 3 columns out of 45 cuts storage read requests and cluster memory consumption dramatically.
8. Exam / Interview Tip Remember: `select()` operates on **columns** (projection), while `filter()` / `where()` operates on **rows** (selection).
9. Key Takeaways - `select()` projects a subset of DataFrame columns. - Enables Catalyst column projection pushdown in columnar formats. - Always prune unneeded columns early in your pipelines.
10. Related Tutorials & Resources - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api) - [PySpark DataFrame Basics (Part 1)](/tutorials/pyspark/pyspark-dataframe-basics)`
Question 22
## Question 22 A retail data pipeline needs to isolate high-value orders by retaining only those transaction records where `order_amount > 500`. Which PySpark DataFrame transformation filters rows based on a given SQL boolean condition?
A. select()
B. filter() (or where())
C. withColumn()
D. distinct()
Answer: ✅ B. filter() (or where())
Explanation:
1. Concept Overview Row-level filtering in PySpark is performed using the **filter()** transformation (or its SQL alias **where()**). It evaluates a boolean expression for each record, passing through only those rows where the condition evaluates to `True`.
2. Why the Correct Option is Right **filter()** (e.g., `df.filter(col("order_amount") > 500)` or `df.where("order_amount > 500")`) filters the rows of the DataFrame. In PySpark, `filter()` and `where()` are 100% identical aliases for the same underlying method.
3. Why Other Options Are Incorrect - **select():** Selects and transforms columns; it does not filter rows based on conditions. - **withColumn():** Adds a new column or replaces an existing column; it does not drop rows. - **distinct():** Removes duplicate rows; it does not filter based on value predicates like `> 500`.
4. Production Scenario / Code Example
from pyspark.sql.functions import col
df_orders = spark.read.table("silver.orders")
# Filtering using column expression:
df_high_value = df_orders.filter(col("order_amount") > 500)
# Equivalent using SQL string predicate:
df_high_value_sql = df_orders.where("order_amount > 500 AND order_status = 'COMPLETED'")
# Catalyst pushes 'order_amount > 500' down into Parquet file statistics!5. Real-World Architecture Context When querying partitioned tables or Parquet files with min/max statistics, Spark's Catalyst Optimizer applies **predicate pushdown**. Micro-partitions or files whose maximum `order_amount` is less than 500 are skipped entirely during file reads.
6. Common Pitfalls & Anti-Patterns - **Using Python `and` / `or` instead of bitwise `&` / `|`:** In PySpark column expressions, you must use bitwise operators with parentheses: `(col("amount") > 500) & (col("status") == "COMPLETED")`. Using `and` throws a `ValueError`. - **Filtering after an expensive join:** Filter both DataFrames *before* joining them to minimize the shuffle volume during the join.
7. Performance & Cost Considerations Early filtering reduces the volume of records propagating through downstream stages, preventing executor out-of-memory (OOM) errors.
8. Exam / Interview Tip Remember: In PySpark, `filter()` and `where()` are completely interchangeable aliases. Both compile to the exact same `Filter` logical plan operator.
9. Key Takeaways - `filter()` filters rows based on boolean conditions. - `where()` is an exact alias for `filter()`. - Always combine conditions using bitwise operators `&` and `|` wrapped in parentheses.
10. Related Tutorials & Resources - [Data Filtering in PySpark DataFrames](/tutorials/pyspark/pyspark-filtering) - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api)`
Question 23
## Question 23 A banking data engineer needs to add a new column named `annual_interest` calculated as `balance * 0.05` to an existing customer accounts DataFrame. Which PySpark DataFrame method is specifically designed to add a new column or replace an existing column by applying an expression?
A. addColumn()
B. selectExpr()
C. withColumn()
D. alterTable()
Answer: ✅ C. withColumn()
Explanation:
1. Concept Overview The **withColumn()** method in PySpark returns a new DataFrame by adding a new column or replacing an existing column that has the same name. It accepts two arguments: the new column name string, and a `Column` expression.
2. Why the Correct Option is Right **withColumn()** is the standard DataFrame API method for creating calculated columns: `df.withColumn("annual_interest", col("balance") * 0.05)` If `annual_interest` does not exist, it is appended to the DataFrame schema; if it already exists, its values are overwritten with the new expression.
3. Why Other Options Are Incorrect - **addColumn():** Does not exist in the PySpark DataFrame API. - **selectExpr():** Projects and calculates expressions, but replaces the entire schema with only the expressions listed unless all existing columns are explicitly selected. - **alterTable():** A DDL operation modifying metadata in a catalog/metastore; not a DataFrame transformation method.
4. Production Scenario / Code Example
from pyspark.sql.functions import col
df_accounts = spark.read.table("silver.customer_accounts")
# Adding a new calculated column:
df_with_interest = df_accounts.withColumn(
"annual_interest",
col("balance") * 0.05
)
# Updating an existing column (e.g., uppercase customer status):
df_updated = df_with_interest.withColumn(
"status",
col("status").upper()
)5. Real-World Architecture Context `withColumn()` is the primary transformation method for feature engineering and data cleansing in PySpark pipelines (converting timestamps, deriving flags, calculating taxes).
6. Common Pitfalls & Anti-Patterns - **Chaining 50 withColumn calls in a loop:** Each `withColumn()` call generates a new internal projected plan. Chaining dozens in a loop can cause the Catalyst Optimizer to take minutes analyzing the plan or throw `StackOverflowError`. For adding multiple columns at once, use `withColumns()` (PySpark 3.3+) or `select()`. - **Passing a raw Python value instead of a Column:** Writing `df.withColumn("tax", 0.05)` fails. Use `lit(0.05)` for literal values.
7. Performance & Cost Considerations `withColumn()` is a **narrow transformation**: it operates within the existing partition without shuffling data across the cluster.
8. Exam / Interview Tip Remember: - Adding a literal value: must wrap in `lit()`, e.g., `withColumn("flag", lit(1))`. - Replacing an existing column: pass the existing column name as the first argument.
9. Key Takeaways - `withColumn(colName, colExpr)` adds or replaces a column. - It is a narrow transformation with zero data shuffle. - Use `lit()` when adding constant values.
10. Related Tutorials & Resources - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api) - [PySpark DataFrame Basics (Part 2)](/tutorials/pyspark/pyspark-dataframe-basics2)`
Question 24
## Question 24 A healthcare analytics pipeline receives clinical test records where identical patient encounter rows have been duplicated due to multiple network retries from hospital telemetry devices. Which PySpark transformation returns a new DataFrame with duplicate rows removed across all columns?
A. distinct() (or dropDuplicates())
B. unique()
C. filterDuplicates()
D. groupBy().count()
Answer: ✅ A. distinct() (or dropDuplicates())
Explanation:
1. Concept Overview Duplicate elimination is a core data hygiene requirement. PySpark provides **distinct()** and **dropDuplicates()** to identify and eliminate duplicate rows from a DataFrame.
2. Why the Correct Option is Right **distinct()** evaluates all columns in the DataFrame and returns a new DataFrame containing only unique rows. Under the hood, `df.distinct()` is an alias for `df.dropDuplicates()` with no arguments specified.
3. Why Other Options Are Incorrect - **unique():** Pandas DataFrame method; does not exist in PySpark DataFrame API. - **filterDuplicates():** Fabricated function name not present in Spark. - **groupBy().count():** Computes aggregation counts for groups; it does not simply deduplicate rows while retaining original schema.
4. Production Scenario / Code Example
df_raw_encounters = spark.read.parquet("s3://hospital-data/encounters/")
# Remove complete duplicate rows across all columns:
df_unique = df_raw_encounters.distinct()
# Or deduplicate based on specific key columns (e.g., patient_id and encounter_time):
df_keyed_unique = df_raw_encounters.dropDuplicates(["patient_id", "encounter_time"])5. Real-World Architecture Context In the Medallion Architecture, deduplication is a primary responsibility of the **Silver layer**. Ingested Bronze data retains all duplicate raw events for auditability, while Silver cleanses duplicates to ensure business aggregates are mathematically sound.
6. Common Pitfalls & Anti-Patterns - **Not realizing distinct() is a wide transformation:** Deduplicating requires comparing rows across the entire dataset, triggering a **Shuffle** across worker nodes. On multi-terabyte datasets, deduplicating unnecessarily consumes significant I/O and network bandwidth. - **Using distinct() when dropDuplicates(subset) is needed:** If two rows have minor differences in an insignificant column (like ingestion timestamp), `distinct()` will consider them different. Use `dropDuplicates(["business_key"])` instead.
7. Performance & Cost Considerations Deduplicating after filtering reduces the volume of records shuffled across executors, saving network bandwidth and execution time.
8. Exam / Interview Tip Compare: - `distinct()`: Deduplicates based on ALL columns. - `dropDuplicates(subset)`: Deduplicates based on a SPECIFIED subset of columns.
9. Key Takeaways - `distinct()` removes duplicate rows across all columns. - Triggers a wide transformation (shuffle). - Use `dropDuplicates(subset)` for business key deduplication.
10. Related Tutorials & Resources - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api) - [Shuffle, Narrow vs Wide Transformations](/tutorials/pyspark/shuffle-narrow-wide-pyspark)`
Question 25
## Question 25 A telecom company wants to sort customer accounts from highest monthly usage to lowest so that executive account managers can identify their top data consumers. Which PySpark transformation and sorting order correctly sorts the DataFrame in descending order?
A. groupBy("monthly_usage").desc()
B. select("monthly_usage").reverse()
C. filter("monthly_usage DESC")
D. orderBy(col("monthly_usage").desc()) (or sort(desc("monthly_usage")))
Answer: ✅ D. orderBy(col("monthly_usage").desc()) (or sort(desc("monthly_usage")))
Explanation:
1. Concept Overview Sorting data in PySpark is performed using **orderBy()** or its identical alias **sort()**. By default, sorting is ascending. To sort from highest to lowest, the column must be qualified with the **desc()** modifier.
2. Why the Correct Option is Right **`orderBy(col("monthly_usage").desc())`** sorts records by the specified column in descending order. In PySpark, `orderBy()` and `sort()` are synonymous, producing a globally ordered DataFrame across all partitions.
3. Why Other Options Are Incorrect - `groupBy("monthly_usage").desc()`: `groupBy()` groups data for aggregation; it does not take sort order arguments. - `select("monthly_usage").reverse()`: `select()` projects columns; `.reverse()` is not a valid DataFrame column sorting method. - `filter("monthly_usage DESC")`: `filter()` evaluates boolean criteria; `DESC` is not a boolean predicate.
4. Production Scenario / Code Example
from pyspark.sql.functions import col, desc
df_usage = spark.read.table("silver.telecom_usage")
# Sorting by monthly usage descending:
df_top_users = df_usage.orderBy(col("monthly_usage").desc())
# Alternatively:
df_top_users_alias = df_usage.sort(desc("monthly_usage"))5. Real-World Architecture Context Global sorting across a distributed dataset requires a Range Partitioner and a full shuffle, meaning every executor redistributes records so that Partition 0 contains the highest values, Partition 1 contains the next tier, and so on.
6. Common Pitfalls & Anti-Patterns - **Sorting intermediate datasets unnecessarily:** Sorting a 10 TB dataset right before a join or aggregation wastes immense compute, because subsequent wide transformations destroy the sort order anyway! Only sort at the final output boundary or when preparing a top-N limit. - **Confusing sort() with sortWithinPartitions():** `sortWithinPartitions()` sorts data locally inside each partition with zero shuffle, but does not provide global sorting.
7. Performance & Cost Considerations Pairing `orderBy(...).desc()` with `limit(N)` is optimized by Spark Catalyst via a **TakeOrderedAndProject** physical operator, which avoids a full global shuffle by maintaining a top-N priority queue per partition.
8. Exam / Interview Tip Remember: `orderBy()` and `sort()` are identical in PySpark. For descending order, use `col("col_name").desc()` or `desc("col_name")`.
9. Key Takeaways - `orderBy()` / `sort()` sorts DataFrames globally. - Use `.desc()` for descending sort order. - Global sorting is a wide transformation; avoid sorting intermediate tables.
10. Related Tutorials & Resources - [Sorting, Sampling & Limit in PySpark](/tutorials/pyspark/sorting-sampling-limit-pyspark) - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api)`
Question 26
## Question 26 A newly joined Data Engineer discovers that an upstream CSV feed includes a 20 MB temporary debugging column named `temp_debug_dump` that is not needed for production analytics. Which PySpark DataFrame method removes this unwanted column from the DataFrame schema?
A. delete()
B. drop()
C. remove()
D. truncate()
Answer: ✅ B. drop()
Explanation:
1. Concept Overview The **drop()** transformation in PySpark returns a new DataFrame with specified columns omitted from the schema. It is the inverse of `withColumn()`.
2. Why the Correct Option is Right **drop()** cleanly removes one or more columns: `df.drop("temp_debug_dump")` The returned DataFrame schema contains all original columns except `temp_debug_dump`.
3. Why Other Options Are Incorrect - **delete():** Does not exist in PySpark DataFrame API; DataFrames are immutable. - **remove():** Python list method; does not exist on PySpark DataFrames. - **truncate():** SQL DDL operation for emptying a table; not a column-dropping DataFrame transformation.
4. Production Scenario / Code Example
df_raw = spark.read.csv("s3://raw-drops/transactions.csv", header=True)
# Drop a single column:
df_clean = df_raw.drop("temp_debug_dump")
# Drop multiple columns at once:
df_final = df_raw.drop("temp_debug_dump", "internal_hash", "legacy_code")5. Real-World Architecture Context Dropping large, unused columns early in the ingestion pipeline prevents memory bloat in executor caches and avoids transmitting unnecessary bytes across shuffle stages.
6. Common Pitfalls & Anti-Patterns - **Expecting drop() to throw an error if the column doesn't exist:** In PySpark, calling `df.drop("non_existent_col")` executes silently and does nothing without raising an exception. - **Attempting to modify the DataFrame in place:** Writing `df.drop("col")` without reassigning `df = df.drop("col")` has zero effect because PySpark DataFrames are strictly immutable!
7. Performance & Cost Considerations `drop()` is a narrow metadata transformation requiring zero data movement or computation until an action is triggered.
8. Exam / Interview Tip Remember: DataFrames are **immutable**. Methods like `drop()`, `withColumn()`, and `select()` always return a *new* DataFrame; they never mutate the original object.
9. Key Takeaways - `drop("col_name")` removes columns from a DataFrame. - Supports dropping multiple columns: `df.drop("c1", "c2")`. - Silently ignores non-existent column names.
10. Related Tutorials & Resources - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api) - [PySpark DataFrame Basics (Part 1)](/tutorials/pyspark/pyspark-dataframe-basics)`
Question 27
## Question 27 A retail company stores product prices as strings (e.g., `"49.99"`) in raw CSV files. Before performing mathematical calculations (such as calculating tax or discounts), the column must be converted to a numeric data type. Which PySpark Column method is used to convert the data type of a column?
A. convert()
B. toType()
C. cast()
D. parseNumeric()
Answer: ✅ C. cast()
Explanation:
1. Concept Overview Data type conversion in PySpark is performed on Column objects using the **cast()** method. It accepts either a string data type representation (e.g., `"decimal(10,2)"`, `"integer"`, `"double"`) or a DataType object from `pyspark.sql.types`.
2. Why the Correct Option is Right **cast()** is the official PySpark method for type casting: `df.withColumn("price", col("price").cast("decimal(10,2)"))` or using the DataType class: `df.withColumn("price", col("price").cast(DoubleType()))` It converts the underlying binary representation to the requested numeric format.
3. Why Other Options Are Incorrect - **convert():** SQL Server function; does not exist on PySpark Column objects. - **toType():** Fabricated function name; not present in Spark. - **parseNumeric():** Non-existent function in PySpark.
4. Production Scenario / Code Example
from pyspark.sql.functions import col
from pyspark.sql.types import DecimalType, IntegerType
df_raw = spark.read.csv("s3://retail/products.csv", header=True)
# Casting string columns to proper numeric types:
df_typed = df_raw .withColumn("price", col("price").cast("decimal(10,2)")) .withColumn("quantity", col("quantity").cast(IntegerType()))
# Inspect updated schema:
df_typed.printSchema()5. Real-World Architecture Context Raw text files (CSV, JSON) often infer string types for all fields during schema-on-read. Casting columns to strict native types (DECIMAL, INT, DATE) in the Silver layer enforces schema integrity and prevents floating-point rounding errors in financial reports.
6. Common Pitfalls & Anti-Patterns - **Unchecked null conversion:** If a string contains invalid characters (e.g., `"N/A"` or `"free"`), `cast("double")` silently converts that cell to `NULL`. Always inspect for unexpected nulls after casting! - **Using Float/Double for financial currency:** Always cast monetary figures to `DecimalType(precision, scale)` rather than Double to avoid floating-point inaccuracies.
7. Performance & Cost Considerations Numeric data types compress much more efficiently in Parquet than raw string representations, shrinking disk storage and accelerating filter scans.
8. Exam / Interview Tip Both string syntax and Type class syntax are valid: - `col("id").cast("int")` - `col("id").cast(IntegerType())`
9. Key Takeaways - `cast()` converts column data types in PySpark. - Unparseable values are silently converted to `NULL`. - Use `DecimalType` for monetary and financial metrics.
10. Related Tutorials & Resources - [PySpark DataFrame Basics (Part 2)](/tutorials/pyspark/pyspark-dataframe-basics2) - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api)`
Question 28
## Question 28 An insurance analytics pipeline needs to classify policyholders based on their age: - If `age < 25`, category is `'YOUTH'` - If `age >= 65`, category is `'SENIOR'` - Otherwise, category is `'STANDARD'` Which PySpark function pair implements this conditional `CASE WHEN ... THEN ... ELSE` branching logic?
A. when() and otherwise()
B. if() and else_if()
C. switch() and default()
D. case() and then()
Answer: ✅ A. when() and otherwise()
Explanation:
1. Concept Overview Conditional branching in PySpark is implemented using **when()** paired with **otherwise()** from the `pyspark.sql.functions` module. This matches the ANSI SQL `CASE WHEN condition THEN result ELSE default END` expression.
2. Why the Correct Option is Right **when() and otherwise()** provide chained conditional logic: `when(condition1, value1).when(condition2, value2).otherwise(default_value)` In this scenario: `when(col("age") < 25, "YOUTH").when(col("age") >= 65, "SENIOR").otherwise("STANDARD")` evaluates conditions sequentially and assigns the appropriate label.
3. Why Other Options Are Incorrect - **if() and else_if():** Python procedural keywords; not Column expression functions for distributed DataFrames. - **switch() and default():** Java/C++ control flow statements; not supported in PySpark. - **case() and then():** ANSI SQL keywords; in PySpark DataFrame API, the functions are named `when()` and `otherwise()`.
4. Production Scenario / Code Example
from pyspark.sql.functions import col, when
df_policyholders = spark.read.table("silver.policyholders")
# Conditional classification using when() and otherwise():
df_classified = df_policyholders.withColumn(
"age_category",
when(col("age") < 25, "YOUTH")
.when(col("age") >= 65, "SENIOR")
.otherwise("STANDARD")
)5. Real-World Architecture Context Using `when() / otherwise()` compiles directly into native Spark Catalyst expressions that execute in vectorized Tungsten CPU code. This is hundreds of times faster than writing a custom Python UDF with Python `if/else` statements!
6. Common Pitfalls & Anti-Patterns - **Writing Python UDFs for simple if/else logic:** Wrapping `if age < 25: return 'YOUTH'` in a Python UDF incurs Python-to-JVM serialization overhead. Always prefer native `when() / otherwise()`! - **Omitting `otherwise()`:** If `otherwise()` is omitted, any row that does not match any `when()` condition is assigned `NULL`.
7. Performance & Cost Considerations Native expressions execute within JVM memory without context switching to the Python runtime, minimizing CPU consumption.
8. Exam / Interview Tip Remember: Multiple conditions are chained with successive `.when()` calls, ending with a single `.otherwise()`.
9. Key Takeaways - `when(condition, value)` evaluates conditional logic. - Chain multiple conditions: `when(...).when(...).otherwise(...)`. - Translates directly to SQL `CASE WHEN` with native vectorized execution.
10. Related Tutorials & Resources - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api) - [PySpark Functions & UDFs Guide](/tutorials/pyspark/pyspark-functions-udfs)`
Question 29
## Question 29 A logistics analytics query must process only shipment records where the shipment status is explicitly `'DELIVERED'`. Which DataFrame operation produces a new DataFrame containing only rows meeting this requirement?
A. select(col("status") == "DELIVERED")
B. drop(col("status") != "DELIVERED")
C. groupBy("status").having("DELIVERED")
D. filter(col("status") == "DELIVERED")
Answer: ✅ D. filter(col("status") == "DELIVERED")
Explanation:
1. Concept Overview Selecting a subset of records based on value criteria is the core purpose of **filtering**. In PySpark, row filtering is executed using `filter()` or `where()`.
2. Why the Correct Option is Right **`filter(col("status") == "DELIVERED")`** passes each record through the equality condition, retaining only rows where `status` is equal to `'DELIVERED'`. Notice the equality operator in Python Column expressions is `==`.
3. Why Other Options Are Incorrect - `select(col("status") == "DELIVERED")`: Evaluates the boolean condition and returns a single column of `True` and `False` booleans; it does not filter rows. - `drop(col("status") != "DELIVERED")`: `drop()` takes column names to remove from the schema; it cannot accept a boolean row-filtering expression. - `groupBy("status").having(...)`: `having()` is not a valid DataFrame method in this manner and groupBy aggregates data.
4. Production Scenario / Code Example
from pyspark.sql.functions import col
df_shipments = spark.read.table("silver.shipments")
# Filter for delivered shipments:
df_delivered = df_shipments.filter(col("status") == "DELIVERED")
# Equivalent SQL syntax:
df_delivered_sql = df_shipments.where("status = 'DELIVERED'")5. Real-World Architecture Context In columnar storage (Parquet/Delta), string columns like `status` benefit from dictionary encoding and file statistics. Filtering early on `status == 'DELIVERED'` allows Spark to skip entire files that contain no delivered shipments.
6. Common Pitfalls & Anti-Patterns - **Using single `=` in Python expressions:** Writing `col("status") = "DELIVERED"` is a Python syntax error. Always use double equals: `==`. - **Handling NULL values:** In SQL and PySpark, `NULL == 'DELIVERED'` evaluates to NULL (not True), so records with NULL status are excluded. If you need nulls, handle them explicitly with `col("status").isNull()`.
7. Performance & Cost Considerations Pushdown filters eliminate unnecessary data before costly joins or aggregations, reducing cluster memory pressure.
8. Exam / Interview Tip Both syntax styles are tested: - Pythonic: `df.filter(col("status") == "DELIVERED")` - SQL string: `df.filter("status = 'DELIVERED'")`
9. Key Takeaways - `filter()` retains rows matching the boolean condition. - Use `==` for equality in column expressions. - Pushes predicates down to storage readers for file skipping.
10. Related Tutorials & Resources - [Data Filtering in PySpark DataFrames](/tutorials/pyspark/pyspark-filtering) - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api)`
Question 30
## Question 30 A legacy mainframe export provides customer data where the customer phone number column is named with an obscure internal abbreviation: `CUST_TEL_NO_PRIM`. The engineering team wants to rename this column to `primary_phone` without modifying any other columns or altering schema order. Which PySpark DataFrame method renames a single column cleanly?
A. renameColumn("CUST_TEL_NO_PRIM", "primary_phone")
B. withColumnRenamed("CUST_TEL_NO_PRIM", "primary_phone")
C. alterColumnName("CUST_TEL_NO_PRIM", "primary_phone")
D. setColumnName("CUST_TEL_NO_PRIM", "primary_phone")
Answer: ✅ B. withColumnRenamed("CUST_TEL_NO_PRIM", "primary_phone")
Explanation:
1. Concept Overview Renaming columns is a standard data transformation task in data ingestion and harmonization. In PySpark, renaming is performed using **withColumnRenamed()**.
2. Why the Correct Option is Right **withColumnRenamed(existingName, newName)** renames the specified column while preserving all other columns, data types, and ordering in the DataFrame schema: `df.withColumnRenamed("CUST_TEL_NO_PRIM", "primary_phone")` It is clean, declarative, and requires no schema redefinition.
3. Why Other Options Are Incorrect - **renameColumn():** Does not exist in PySpark (common confusion with Pandas `rename(columns={...})`). - **alterColumnName():** Does not exist in the DataFrame API. - **setColumnName():** Fabricated function name.
4. Production Scenario / Code Example
df_mainframe = spark.read.parquet("s3://mainframe-drops/customers/")
# Renaming a single column:
df_clean = df_mainframe.withColumnRenamed("CUST_TEL_NO_PRIM", "primary_phone")
# Renaming multiple columns by chaining:
df_standardized = df_mainframe .withColumnRenamed("CUST_TEL_NO_PRIM", "primary_phone") .withColumnRenamed("CUST_ADDR_LN_1", "address_line_1")5. Real-World Architecture Context In enterprise lakehouses, raw Bronze tables preserve original source column names. During Silver transformation, standardized naming conventions (snake_case, meaningful English business terms) are enforced using `withColumnRenamed()`.
6. Common Pitfalls & Anti-Patterns - **Non-existent column names:** If `existingName` does not exist in the DataFrame, PySpark silently does nothing and returns the DataFrame unchanged without throwing an error! - **Chaining 50 withColumnRenamed calls:** For renaming dozens of columns simultaneously, using `toDF(*new_columns)` or a loop over `select([col(c).alias(mapping.get(c, c)) for c in df.columns])` is significantly faster during query planning.
7. Performance & Cost Considerations `withColumnRenamed()` is a pure metadata transformation. It updates the Catalyst logical plan with zero data movement or shuffle.
8. Exam / Interview Tip Remember the order of arguments: `withColumnRenamed(existing_name, new_name)` (Old name first, New name second).
9. Key Takeaways - `withColumnRenamed()` updates a column name in the schema. - Does not modify underlying data or trigger a shuffle. - Silently ignores the operation if the original column name is not found.
10. Related Tutorials & Resources - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api) - [PySpark DataFrame Basics (Part 2)](/tutorials/pyspark/pyspark-dataframe-basics2)`
Question 31
## Question 31 A data engineer applies multiple filters to a 10-million row DataFrame. The engineer needs to print the exact number of rows remaining in the filtered DataFrame to the console. Which PySpark DataFrame Action triggers distributed execution and returns the total row count as an integer?
A. length()
B. size()
C. count()
D. rowCount()
Answer: ✅ C. count()
Explanation:
1. Concept Overview In Apache Spark, operations are divided into **Transformations** (which are lazily evaluated) and **Actions** (which trigger physical execution and return results to the driver program). **count()** is a foundational DataFrame Action that returns the number of rows.
2. Why the Correct Option is Right **count()** triggers Spark execution across all cluster worker nodes. Each executor computes the row count for its local partitions, and the driver aggregates these counts and returns the final integer to the client program: `num_rows = df_filtered.count()`
3. Why Other Options Are Incorrect - **length():** SQL function that returns the character length of a string column; not a DataFrame action. - **size():** Function that returns the length of an array or map column; not a DataFrame action. - **rowCount():** Fabricated function name; does not exist in Spark.
4. Production Scenario / Code Example
df_orders = spark.read.table("silver.orders")
# Apply lazy transformations:
df_filtered = df_orders.filter("order_amount > 1000 AND status = 'COMPLETED'")
# Trigger execution and return row count:
total_high_value_orders = df_filtered.count()
print(f"Total eligible orders: {total_high_value_orders}")5. Real-World Architecture Context Data pipelines frequently use `count()` to log metrics and validate ETL output thresholds (e.g., verifying that today's batch processed at least 90% of yesterday's volume).
6. Common Pitfalls & Anti-Patterns - **Calling count() repeatedly in a pipeline:** Calling `df.count()` multiple times without caching forces Spark to recompute the entire DAG from disk each time! If you need the count and also need to write the DataFrame, consider caching or computing metrics during write. - **Using count() just to check if a DataFrame is empty:** If you only need to verify whether at least one row exists, `df.take(1)` or `df.limit(1).count()` is far faster than counting all 10 million rows!
7. Performance & Cost Considerations On Parquet/Delta tables without filters, `df.count()` is optimized by Catalyst to read metadata statistics directly in milliseconds without scanning data files.
8. Exam / Interview Tip Remember: `count()` is an **ACTION**, not a transformation. Calling `count()` triggers the immediate execution of all upstream transformations.
9. Key Takeaways - `count()` is a distributed Action that returns total row count. - Triggers execution of the Catalyst execution plan. - For checking emptiness, prefer `df.take(1)` over `df.count() > 0`.
10. Related Tutorials & Resources - [Data Aggregation in PySpark DataFrames](/tutorials/pyspark/pyspark-aggregation) - [RDD Basics — Creation, Transformation & Actions](/tutorials/pyspark/rdd-basics)`
Question 32
## Question 32 A data scientist needs to inspect a small sample of a multi-terabyte dataset during exploratory analysis. The scientist wants to restrict the DataFrame to only the first 100 rows. Which PySpark DataFrame transformation restricts the result set to a specified maximum number of rows?
A. limit(100)
B. top(100)
C. head(100)
D. sample(100)
Answer: ✅ A. limit(100)
Explanation:
1. Concept Overview The **limit(n)** transformation returns a new DataFrame containing at most the first `n` rows. It is lazily evaluated and optimized by Spark to avoid reading the entire dataset.
2. Why the Correct Option is Right **limit(100)** creates a new DataFrame restricted to 100 records: `df_sample = df.limit(100)` Spark optimizes `limit()` by reading only the first few partitions until 100 rows are collected, halting execution on remaining partitions to conserve cluster compute.
3. Why Other Options Are Incorrect - **top(100):** An RDD action (not a DataFrame transformation) that returns an array of the top elements in descending order to the driver. - **head(100):** An Action that collects the first 100 rows as a Python list of `Row` objects to the driver; it does not return a new DataFrame. - **sample(100):** Takes a fraction between 0.0 and 1.0 (e.g., `sample(fraction=0.1)`), not an absolute row count.
4. Production Scenario / Code Example
df_large = spark.read.table("gold.massive_clickstream")
# Transformation: Returns a new DataFrame limited to 100 records
df_test = df_large.limit(100)
# Write test subset to scratch location:
df_test.write.parquet("s3://scratch/test_subset/")5. Real-World Architecture Context In exploratory data analysis (EDA) in Databricks notebooks or Jupyter, developers use `limit()` before displaying tables or exporting samples to local environments, preventing driver memory crashes.
6. Common Pitfalls & Anti-Patterns - **Expecting deterministic rows without sorting:** Calling `df.limit(100)` without an `orderBy()` returns arbitrary rows depending on partition arrival order. For deterministic output, use `df.orderBy("id").limit(100)`. - **Confusing limit() with show():** `limit()` is a transformation returning a DataFrame; `show()` is an action printing an ASCII table to the console.
7. Performance & Cost Considerations Spark pushes limit operations down into file readers. If 100 rows are found in the first Parquet file, Spark skips reading the remaining 10,000 files in the dataset!
8. Exam / Interview Tip Compare: - `limit(n)`: **Transformation** returning a DataFrame with at most `n` rows. - `take(n)` / `head(n)`: **Action** returning a Python list of `n` `Row` objects to the driver.
9. Key Takeaways - `limit(n)` returns a new DataFrame capped at `n` rows. - Highly optimized: halts scanning once `n` records are found. - Pair with `orderBy()` for deterministic top-N subsets.
10. Related Tutorials & Resources - [Sorting, Sampling & Limit in PySpark](/tutorials/pyspark/sorting-sampling-limit-pyspark) - [PySpark DataFrame Basics (Part 1)](/tutorials/pyspark/pyspark-dataframe-basics)`
Question 33
## Question 33 A company has two regional sales DataFrames: `df_us_sales` and `df_eu_sales`. Both DataFrames have identical column structures and data types. Which PySpark DataFrame method combines the rows of both DataFrames into a single unified DataFrame, equivalent to SQL `UNION ALL`?
A. join()
B. merge()
C. concat()
D. union() (or unionAll())
Answer: ✅ D. union() (or unionAll())
Explanation:
1. Concept Overview Row concatenation across DataFrames with matching schemas is performed using the **union()** method. In PySpark, `union()` resolves by column position and preserves duplicate rows (behaving like SQL `UNION ALL`).
2. Why the Correct Option is Right **union()** (or its legacy alias `unionAll()`) merges two DataFrames vertically: `df_global = df_us_sales.union(df_eu_sales)` It stacks the rows of both DataFrames into a single combined DataFrame without removing duplicates.
3. Why Other Options Are Incorrect - **join():** Combines DataFrames horizontally based on a join condition (key matching); it does not stack rows vertically. - **merge():** Pandas method; does not exist on PySpark DataFrames. - **concat():** PySpark function that concatenates string columns together horizontally (e.g., first_name + last_name); it does not combine DataFrames.
4. Production Scenario / Code Example
df_us = spark.read.table("silver.us_sales")
df_eu = spark.read.table("silver.eu_sales")
# Combine rows using union (SQL UNION ALL behavior):
df_global = df_us.union(df_eu)
# If column orders differ, use unionByName to match by column name instead of position:
df_global_safe = df_us.unionByName(df_eu)5. Real-World Architecture Context In global data lakehouses, transactions land in regional silos. The Silver layer standardizes regional schemas and applies `unionByName()` to assemble global enterprise datasets.
6. Common Pitfalls & Anti-Patterns - **Assuming union() deduplicates rows:** Standard SQL `UNION` removes duplicates, but in PySpark, `union()` preserves duplicates (like `UNION ALL`). To deduplicate after a union, explicitly call `.distinct()`. - **Positional mismatch with union():** Standard `union()` matches columns by index position ($1, $2, $3), not by name! If `df1` has `(id, name)` and `df2` has `(name, id)`, `union()` corrupts data. Always prefer **`unionByName()`**!
7. Performance & Cost Considerations `union()` is a **narrow transformation** that requires zero data shuffling between executors, executing with near-instantaneous speed.
8. Exam / Interview Tip Remember: In PySpark, `union()` does NOT remove duplicates. It is equivalent to SQL **UNION ALL**. To match columns safely by name, use **`unionByName()`**.
9. Key Takeaways - `union()` stacks DataFrames vertically (SQL `UNION ALL`). - Does NOT remove duplicate rows. - Use `unionByName()` to guard against mismatched column positions.
10. Related Tutorials & Resources - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api) - [PySpark DataFrame Basics (Part 2)](/tutorials/pyspark/pyspark-dataframe-basics2)`
Question 34
## Question 34 A clinical research organization needs to compute summary statistics: the average patient age and total number of clinical visits grouped by hospital department. Which distributed data processing concept describes combining multiple individual records into summarized metric values?
A. Partition Pruning
B. Data Aggregation (groupBy and aggregate functions)
C. Schema Evolution
D. Broadcast Distribution
Answer: ✅ B. Data Aggregation (groupBy and aggregate functions)
Explanation:
1. Concept Overview **Data Aggregation** is the process of grouping individual transaction rows and calculating summary metrics (e.g., `AVG`, `SUM`, `COUNT`, `MAX`, `MIN`) across those groups. In PySpark, aggregation is performed using `groupBy()` combined with aggregate functions or the `agg()` method.
2. Why the Correct Option is Right **Data Aggregation (groupBy and aggregate functions)** precisely defines this operation: `df.groupBy("department").agg(avg("age"), count("visit_id"))` It groups records by department and reduces thousands of individual patient rows down to a concise summary table of metrics.
3. Why Other Options Are Incorrect - **Partition Pruning:** Storage read optimization that skips non-relevant file directories during query planning. - **Schema Evolution:** Ability of table formats (Delta Lake) to adapt to new columns added over time. - **Broadcast Distribution:** Mechanism for transmitting a small lookup table to all worker nodes to avoid shuffle joins.
4. Production Scenario / Code Example
from pyspark.sql.functions import avg, count
df_patients = spark.read.table("silver.patient_visits")
# Data Aggregation:
df_summary = df_patients.groupBy("department").agg(
avg("patient_age").alias("avg_age"),
count("visit_id").alias("total_visits")
)
df_summary.show()5. Real-World Architecture Context In the Medallion Architecture, aggregation is the primary hallmark of the **Gold layer**. Raw records in Silver are aggregated into curated business metrics, KPIs, and dimensional models for executive consumption.
6. Common Pitfalls & Anti-Patterns - **Data skew during aggregation:** If 90% of visits belong to the "General Medicine" department, a single executor will be overwhelmed processing that group while other workers sit idle. - **Aggregating before cleaning:** Calculating averages before filtering out invalid dummy records (e.g., age = 999) skews metrics.
7. Performance & Cost Considerations Spark optimizes aggregations using a two-stage process: partial aggregation locally on workers before shuffling (map-side combine), minimizing the volume of data transmitted over the network.
8. Exam / Interview Tip Remember: Aggregations are **wide transformations** that trigger a cluster shuffle. Spark uses `spark.sql.shuffle.partitions` (default 200) to control post-aggregation partition counts.
9. Key Takeaways - Aggregation collapses detail rows into summary metrics. - Implemented via `groupBy().agg()`. - Core transformation for Gold-layer analytics and BI marts.
10. Related Tutorials & Resources - [Data Aggregation in PySpark DataFrames](/tutorials/pyspark/pyspark-aggregation) - [Shuffle, Narrow vs Wide Transformations](/tutorials/pyspark/shuffle-narrow-wide-pyspark)`
Question 35
## Question 35 A banking analytics platform needs to determine the exact count of unique customers who completed transactions in the past month. Which PySpark code pattern correctly calculates the number of distinct customer IDs?
A. df.groupBy("customer_id").count()
B. df.filter("customer_id").size()
C. df.select("customer_id").distinct().count() (or df.select(countDistinct("customer_id")))
D. df.describe("customer_id").max()
Answer: ✅ C. df.select("customer_id").distinct().count() (or df.select(countDistinct("customer_id")))
Explanation:
1. Concept Overview Calculating the cardinality of unique values in a column can be accomplished in PySpark using two standard approaches: 1. Transforming the column and counting unique rows: `df.select("customer_id").distinct().count()`. 2. Using the built-in aggregate function: `df.select(countDistinct("customer_id"))`.
2. Why the Correct Option is Right **`df.select("customer_id").distinct().count()`** isolates the `customer_id` column, eliminates duplicate IDs across all records using `distinct()`, and executes `count()` to return the exact integer count of unique customers.
3. Why Other Options Are Incorrect - `df.groupBy("customer_id").count()`: Groups by customer and returns a DataFrame listing every individual customer ID with their transaction count; it does not return the *total count of unique customers*. - `df.filter("customer_id").size()`: Invalid syntax; `filter()` requires a boolean condition, and `.size()` is not a DataFrame action. - `df.describe("customer_id").max()`: Computes descriptive summary statistics (min, max, mean) as strings, not unique entity cardinality.
4. Production Scenario / Code Example
from pyspark.sql.functions import countDistinct, col
df_tx = spark.read.table("silver.bank_transactions")
# Method 1: DataFrame transformation + action (returns integer directly to driver)
unique_customers = df_tx.select("customer_id").distinct().count()
print(f"Total Unique Customers: {unique_customers}")
# Method 2: Using countDistinct aggregate function (returns single-cell DataFrame)
df_metric = df_tx.select(countDistinct("customer_id").alias("unique_cust_count"))5. Real-World Architecture Context Calculating distinct counts is standard in customer churn, monthly active user (MAU), and marketing reach metrics.
6. Common Pitfalls & Anti-Patterns - **Exact countDistinct on billions of records:** Exact distinct counting requires hashing and shuffling every key, which is resource-intensive. If an approximate count (+/- 2% error) is acceptable for large-scale BI, use **`approx_count_distinct("customer_id")`** (HyperLogLog) to speed up execution by 10x! - **Not filtering out NULLs:** `countDistinct()` ignores NULLs, whereas `distinct().count()` counts `NULL` as one unique value!
7. Performance & Cost Considerations For massive datasets, `approx_count_distinct()` uses fixed memory and minimal shuffle, delivering massive cost and runtime savings.
8. Exam / Interview Tip - Exact unique count: `countDistinct("col")` or `select("col").distinct().count()`. - Approximate unique count (HyperLogLog): `approx_count_distinct("col")`.
9. Key Takeaways - Calculate unique entities using `distinct().count()` or `countDistinct()`. - `distinct()` is a wide transformation (triggers shuffle). - Use `approx_count_distinct()` for petabyte-scale approximate metrics.
10. Related Tutorials & Resources - [Data Aggregation in PySpark DataFrames](/tutorials/pyspark/pyspark-aggregation) - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api)`
Question 36
## Question 36 A manufacturing data engineer writes a PySpark script containing: `df1 = spark.read.csv("s3://sensors/")` `df2 = df1.filter("temp > 100")` `df3 = df2.withColumn("alert", lit(True))` `df4 = df3.select("sensor_id", "alert")` During execution, when does Apache Spark ACTUALLY read the CSV files from S3 and execute these transformations?
A. Only when an Action (such as .write, .count(), or .collect()) is invoked on the DataFrame
B. Immediately on line 1 when spark.read.csv() is defined
C. On line 2 when the filter condition is applied
D. Whenever the Python garbage collector runs
Answer: ✅ A. Only when an Action (such as .write, .count(), or .collect()) is invoked on the DataFrame
Explanation:
1. Concept Overview Apache Spark employs **Lazy Evaluation**. Transformations (e.g., `read`, `filter`, `withColumn`, `select`) do not execute immediately when defined in code. Instead, Spark records them as a Directed Acyclic Graph (DAG) of logical instructions in the Catalyst Optimizer.
2. Why the Correct Option is Right Data is read and processed **only when an Action is invoked** (such as `.write.save()`, `.count()`, `.show()`, `.collect()`, or `.take()`). The invocation of an Action triggers Spark to compile the logical plan, optimize it (pruning unneeded columns and pushing down filters), and submit physical execution stages to cluster worker nodes.
3. Why Other Options Are Incorrect - **Immediately on line 1:** `spark.read.csv()` only checks schema metadata; it does not read the full data files. - **On line 2 when filter is applied:** `filter()` is a transformation; it merely appends a Filter node to the unresolved logical plan. - **When garbage collector runs:** Python garbage collection frees local memory; it has no role in triggering distributed Spark cluster execution.
4. Production Scenario / Code Example
from pyspark.sql.functions import lit
# Lines 1-4: PURE TRANSFORMATIONS (Zero data read, zero network I/O, runs in 2 milliseconds!)
df1 = spark.read.csv("s3://sensors/", header=True)
df2 = df1.filter("temp > 100")
df3 = df2.withColumn("alert", lit(True))
df4 = df3.select("sensor_id", "alert")
# Line 5: ACTION! (Triggers physical execution across all worker nodes)
df4.write.mode("overwrite").parquet("s3://alerts/")5. Real-World Architecture Context Lazy evaluation is Spark's core optimization superpower. Because Spark waits until an action is called, the Catalyst Optimizer can see the entire end-to-end pipeline before running. It notices that only `sensor_id` is needed (pruning all other columns) and pushes `temp > 100` down to the storage reader.
6. Common Pitfalls & Anti-Patterns - **Adding .count() or .show() after every line for debugging:** Placing `.show()` after every transformation destroys lazy optimization, forcing Spark to execute multiple separate, un-optimized jobs and slowing pipeline runtime by 10x. - **Catching bugs late:** Errors like column name typos are not detected until an action triggers plan analysis unless eager analysis is enabled.
7. Performance & Cost Considerations Lazy evaluation allows Spark to fuse multiple operations into single vectorized execution stages (Whole-Stage Code Generation), maximizing CPU cache efficiency.
8. Exam / Interview Tip Core distinction to memorize: - **Transformations** (Lazy): `select`, `filter`, `withColumn`, `join`, `groupBy`, `distinct`. - **Actions** (Eager): `count`, `collect`, `show`, `take`, `write`, `save`.
9. Key Takeaways - Spark transformations are evaluated lazily. - Execution occurs only when an Action is called. - Enables Catalyst to optimize queries globally before reading data.
10. Related Tutorials & Resources - [Catalyst Optimizer & Tungsten Engine in PySpark](/tutorials/pyspark/catalyst-tungsten-pyspark) - [RDD Basics — Creation, Transformation & Actions](/tutorials/pyspark/rdd-basics)`
Question 37
## Question 37 A retail analytics team needs to retrieve customer accounts located in either California OR New York whose account status is active: `(state = 'CA' OR state = 'NY') AND status = 'ACTIVE'` Which PySpark DataFrame syntax correctly applies this compound filtering logic using Column expressions?
A. df.filter((col("state") == "CA" or col("state") == "NY") and col("status") == "ACTIVE")
B. df.select("state == 'CA' || state == 'NY' && status == 'ACTIVE'")
C. df.where().or("CA", "NY").and("ACTIVE")
D. df.filter(((col("state") == "CA") | (col("state") == "NY")) & (col("status") == "ACTIVE"))
Answer: ✅ D. df.filter(((col("state") == "CA") | (col("state") == "NY")) & (col("status") == "ACTIVE"))
Explanation:
1. Concept Overview In PySpark, constructing compound boolean filter conditions requires understanding Python's operator overloading rules. Because Python does not allow overriding the logical `and`, `or`, and `not` keywords for custom objects, PySpark overloads the bitwise operators: - Bitwise AND: `&` - Bitwise OR: `|` - Bitwise NOT: `~`
2. Why the Correct Option is Right **`df.filter(((col("state") == "CA") | (col("state") == "NY")) & (col("status") == "ACTIVE"))`** is the correct syntax. Every individual condition must be enclosed in parentheses because bitwise operators (`&`, `|`) have higher operator precedence in Python than comparison operators (`==`).
3. Why Other Options Are Incorrect - `df.filter(... or ... and ...)`: Using Python logical keywords (`and`, `or`) on PySpark Column objects raises a `ValueError: Cannot convert column into bool`. - `df.select(... || ...)`: `select()` projects columns, not row filters, and `||` is not valid Python syntax. - `df.where().or().and()`: Chained methods like `.or()` and `.and()` do not exist in the DataFrame API.
4. Production Scenario / Code Example
from pyspark.sql.functions import col
df_customers = spark.read.table("silver.customers")
# Correct: Bitwise operators with explicit parentheses:
df_target = df_customers.filter(
((col("state") == "CA") | (col("state") == "NY")) &
(col("status") == "ACTIVE")
)
# Alternative: SQL string syntax inside filter (also valid and clean!):
df_target_sql = df_customers.filter("(state = 'CA' OR state = 'NY') AND status = 'ACTIVE'")5. Real-World Architecture Context Writing clean, safe filter expressions prevents subtle production bugs. Enclosing sub-clauses in parentheses ensures logical operators evaluate in the intended business order.
6. Common Pitfalls & Anti-Patterns - **Omitting parentheses around sub-expressions:** Writing `col("state") == "CA" | col("state") == "NY"` evaluates `"CA" | col("state")` first due to Python precedence, throwing a `TypeError`! - **Using single ampersand without parens:** Always wrap each comparison in parentheses: `(A) & (B)`.
7. Performance & Cost Considerations The Catalyst Optimizer simplifies compound boolean predicates into disjunctive normal forms, allowing pushdown of partition filters directly into cloud storage metadata.
8. Exam / Interview Tip Remember the rule for PySpark boolean filters: 1. Use `&` for AND, `|` for OR, `~` for NOT. 2. Every sub-expression MUST be wrapped in parentheses: `(cond1) & (cond2)`.
9. Key Takeaways - Use bitwise operators `&` and `|` for compound filters. - Enclose all comparison expressions in parentheses. - Alternatively, pass a clean SQL string to `filter()`.
10. Related Tutorials & Resources - [Data Filtering in PySpark DataFrames](/tutorials/pyspark/pyspark-filtering) - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api)`
Question 38
## Question 38 A telecom customer churn dataset contains missing values in a numeric column named `monthly_data_gb`. Data science models cannot process nulls and require that all missing values in this column be replaced with `0.0`. Which PySpark DataFrame method cleanly replaces NULL values with a default value?
A. replaceNulls(0.0)
B. fillna(0.0, subset=["monthly_data_gb"]) (or na.fill())
C. dropNulls()
D. cleanMissing(0.0)
Answer: ✅ B. fillna(0.0, subset=["monthly_data_gb"]) (or na.fill())
Explanation:
1. Concept Overview Handling missing data is a fundamental data preparation task. PySpark provides **fillna()** (and its equivalent `df.na.fill()`) to replace NULL / NaN values in DataFrames with specified replacement constants.
2. Why the Correct Option is Right **`fillna(0.0, subset=["monthly_data_gb"])`** replaces NULL values in the specified column with `0.0`. PySpark matches data types automatically, ensuring numeric columns receive numeric defaults without altering other columns.
3. Why Other Options Are Incorrect - **replaceNulls():** Fabricated function name; does not exist in PySpark. - **dropNulls():** Non-existent; the method to drop nulls is `dropna()` or `df.na.drop()`, which removes rows rather than replacing values. - **cleanMissing():** Non-existent function.
4. Production Scenario / Code Example
df_telecom = spark.read.table("silver.telecom_customers")
# Method 1: fillna with subset
df_imputed = df_telecom.fillna(0.0, subset=["monthly_data_gb"])
# Method 2: Dictionary mapping different defaults for multiple columns
df_multi_imputed = df_telecom.fillna({
"monthly_data_gb": 0.0,
"roaming_charges": 0.0,
"plan_tier": "UNKNOWN"
})5. Real-World Architecture Context In machine learning feature preparation and analytical reporting, NULL values cause failures in linear regression algorithms and misrepresent sum aggregations. Replacing nulls at the Silver-to-Gold boundary ensures downstream models train reliably.
6. Common Pitfalls & Anti-Patterns - **Calling fillna without specifying subset:** Calling `df.fillna(0)` without a subset will attempt to replace NULLs across *every* integer column in the DataFrame, which may inadvertently alter columns where NULL has a distinct business meaning. - **Type mismatch:** Passing a string default `df.fillna("N/A")` will only replace NULLs in string/text columns; it will silently ignore numeric columns!
7. Performance & Cost Considerations `fillna()` is a narrow transformation requiring no data shuffle. It translates to an internal `coalesce(col, default)` expression.
8. Exam / Interview Tip Both syntax styles are equivalent: - `df.fillna(value, subset=[...])` - `df.na.fill(value, subset=[...])`
9. Key Takeaways - `fillna()` replaces NULL values with default constants. - Specify `subset=[...]` to target specific columns. - Pass a dictionary to specify different defaults per column.
10. Related Tutorials & Resources - [Handling Missing Data in PySpark DataFrames](/tutorials/pyspark/pyspark-missing) - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api)`
Question 39
## Question 39 A customer dataset imported from legacy flat files contains inconsistent leading and trailing whitespace in customer name strings (e.g., `" Alice Smith "`), causing equality joins and grouping operations to fail. Which PySpark SQL function removes both leading and trailing whitespace from string columns?
A. strip()
B. clean()
C. trim()
D. ltrim()
Answer: ✅ C. trim()
Explanation:
1. Concept Overview String sanitization is essential when ingesting delimited or fixed-width text files. PySpark provides built-in string cleaning functions in `pyspark.sql.functions`: - `trim()`: Removes leading and trailing whitespace. - `ltrim()`: Removes leading whitespace only. - `rtrim()`: Removes trailing whitespace only.
2. Why the Correct Option is Right **trim()** removes spaces from both the beginning and end of a string column: `df.withColumn("customer_name", trim(col("customer_name")))` It sanitizes `" Alice Smith "` into `"Alice Smith"`, ensuring clean equality joins and distinct groupings.
3. Why Other Options Are Incorrect - **strip():** Standard Python string method; not a PySpark Column expression function. - **clean():** Non-existent function in PySpark. - **ltrim():** Left trim only; leaves trailing whitespace intact.
4. Production Scenario / Code Example
from pyspark.sql.functions import col, trim, lower
df_raw = spark.read.csv("s3://landing/customers.csv", header=True)
# Standardizing string columns: trim whitespace and normalize casing
df_clean = df_raw.withColumn(
"customer_name",
trim(col("customer_name"))
).withColumn(
"email",
lower(trim(col("email")))
)5. Real-World Architecture Context In enterprise lakehouses, dirty strings cause downstream joins to drop matches (e.g., `"NY"` vs `"NY "`). Trimming and standardizing string fields in the Silver layer guarantees referential integrity during downstream warehouse joins.
6. Common Pitfalls & Anti-Patterns - **Using Python `.strip()` in a UDF:** Writing a custom Python UDF to call `x.strip()` causes massive Python-JVM serialization penalties. Always use native `trim()`! - **Not trimming join keys:** If join match rates are unexpectedly low, check for hidden trailing whitespace in legacy char/text fields.
7. Performance & Cost Considerations Native `trim()` executes in C++ / Java bytecode using Whole-Stage Code Generation, processing millions of rows per second per core.
8. Exam / Interview Tip Remember: - `trim(col)`: Strips BOTH ends. - `ltrim(col)`: Strips LEFT (leading) only. - `rtrim(col)`: Strips RIGHT (trailing) only.
9. Key Takeaways - `trim()` eliminates leading and trailing whitespace. - Native expression with zero UDF overhead. - Critical for joining and grouping dirty text feeds.
10. Related Tutorials & Resources - [PySpark Functions & UDFs Guide](/tutorials/pyspark/pyspark-functions-udfs) - [DataFrame API — Select, Filter, WithColumn & Drop](/tutorials/pyspark/df-api)`
Question 40
## Question 40 A global retailer processes billions of records daily using PySpark. A junior developer observes that calling `.filter()`, `.select()`, and `.withColumn()` returns a new DataFrame in milliseconds, regardless of whether the dataset is 10 megabytes or 50 terabytes in size. What foundational Apache Spark execution architecture explains why transformations return immediately without processing data?
A. Lazy Evaluation (Spark builds an execution plan and defers computation until an Action is called)
B. Spark writes all data to local executor SSDs immediately upon transformation
C. Spark caches all 50 terabytes of data directly into driver RAM memory
D. Spark uses peer-to-peer torrent networking to process data on client machines
Answer: ✅ A. Lazy Evaluation (Spark builds an execution plan and defers computation until an Action is called)
Explanation:
1. Concept Overview **Lazy Evaluation** is the foundational architectural paradigm of Apache Spark. In Spark, operations on distributed collections (RDDs, DataFrames, Datasets) are split into two categories: 1. **Transformations:** Define what data operations to apply; they are evaluated lazily. 2. **Actions:** Instruct Spark to execute the computation and return a result or write data to storage.
2. Why the Correct Option is Right Under **Lazy Evaluation**, calling transformations like `.filter()`, `.select()`, or `.withColumn()` does not process a single byte of data. Spark simply constructs a logical query plan in memory. The method returns a new DataFrame object wrapping this plan in milliseconds. Computation only begins when an **Action** (e.g., `.write`, `.count()`, `.collect()`) is called.
3. Why Other Options Are Incorrect - **Writes data to SSDs immediately:** Spark does not write data to disk during transformation definitions; disk writes occur during shuffles or explicit file write actions. - **Caches 50 TB into driver RAM:** Driver RAM would crash instantly with an OutOfMemoryError; data resides on distributed workers or storage, not driver memory. - **Peer-to-peer torrent networking:** Fabricated mechanism; Spark uses a master/worker distributed cluster architecture.
4. Production Scenario / Code Example
# A 50 TB dataset
df_massive = spark.read.parquet("s3://massive-lake/transactions/")
# Defining transformations: Execution completes in 3 milliseconds!
df_step1 = df_massive.filter("transaction_year = 2024")
df_step2 = df_step1.withColumn("tax", col("amount") * 0.08)
df_step3 = df_step2.select("customer_id", "tax")
# Nothing has actually been computed yet!
print("Plan created instantly via Lazy Evaluation.")
# ACTION: Physical execution begins now!
df_step3.write.mode("overwrite").parquet("s3://massive-lake/2024_tax/")