1. Concept Overview The `count()` method in PySpark is an **action** that returns the total number of rows present in the DataFrame as an integer. Because the input DataFrame contains three distinct rows with IDs 1, 2, and 3, calling `df.count()` evaluates the execution plan across all partitions and returns the exact integer value `3`.

2. Architecture & Design Principle In Apache Spark's Catalyst engine, operations are divided into **transformations** (lazy) and **actions** (eager): 1. Transformations (e.g., `filter()`, `select()`) build a directed acyclic graph (DAG) of logical execution steps without executing immediately. 2. Actions (e.g., `count()`, `collect()`, `show()`, `write`) trigger the Catalyst optimizer, compile the physical execution plan, and dispatch tasks to executor cores across the cluster. For `count()`, each executor counts the rows in its local partitions and transmits the partial integer sum back to the Spark Driver program, which aggregates them into the final count.

3. Step-by-Step Code / Logic Execution 1. Initialization: A DataFrame `df` is created with 3 rows:

   data = [(1,), (2,), (3,)]
   df = spark.createDataFrame(data, ["id"])

2. When df.count() is invoked: - Spark creates a ShuffleExchange or partition-level count aggregation (partial_count). - Leaf tasks on executor nodes evaluate the number of records (1, 1, 1). - Driver executes the final aggregate: 1 + 1 + 1 = 3. 3. The return value is an integer: 3.

4. Why the Correct Answer Is Optimal Option A is correct because: - The DataFrame has exactly 3 rows, none of which have been filtered out or dropped. - `count()` accurately aggregates every existing record in the dataset, including records with null attributes if they exist.

5. Why Alternative Options Fail - Option B (2): Incorrect; no filter or limit was applied to drop a row. - Option C (1): Incorrect; `count()` computes the full dataset size, not a single row or first element. - Option D (Error): Incorrect; `df.count()` is a valid core DataFrame API method supported across all Spark versions.

6. Edge Cases & Production Pitfalls - **Null Values**: In SQL, `COUNT(column_name)` ignores NULL values, whereas `df.count()` in PySpark counts all rows regardless of whether columns contain NULL values (equivalent to SQL `COUNT(*)`). - **Expensive Computation**: On multi-terabyte datasets, calling `df.count()` forces a full scan of all underlying data files. Calling `df.count()` multiple times in a script causes redundant scans unless the DataFrame is cached.

7. Performance & Optimization Considerations - In Delta Lake or Parquet metadata-rich formats, Spark can occasionally resolve `count()` directly from file transaction log metadata without scanning underlying Parquet data files. - To check if a DataFrame is empty, avoid `if df.count() > 0:` (which scans the entire table); use `if not df.isEmpty():` or `if len(df.take(1)) > 0:` instead to terminate after reading the first record.

8. Reliability & Operational Best Practices - Never use `df.count()` inside tight loops or before every transformation step in production pipelines, as each call submits a separate Spark job. - Use Spark UI (**Jobs** tab) to verify the number of stages generated by `count()`.

9. Real-World Architectural Scenario An ad-tech telemetry pipeline processes 15 billion click events per day. A junior engineer added `df.count()` logging statements after every cleaning step to track record counts, increasing job execution time from 20 minutes to 2.5 hours. Replacing redundant `count()` calls with an end-of-stage accumulator slashed cluster compute costs by 85%.

10. Official Documentation & Learning Resources - Core PySpark Concepts: [PySpark DataFrame Basics Tutorial](/tutorials/pyspark/pyspark-dataframe-basics)

1. Concept Overview The `distinct()` transformation in PySpark removes duplicate rows across the selected columns. In the input dataset, the `country` column contains three records: "India", "India", and "USA". Calling `distinct()` collapses the duplicate occurrences of "India" into a single unique record, leaving "India" and "USA". Calling `count()` on the resulting DataFrame yields `2`.

2. Architecture & Design Principle 1. **Wide Transformation (Shuffle)**: In a distributed system, identical records may reside on different worker nodes. To identify and deduplicate records, Spark performs a **hash partition shuffle**: - Each row's key (here, `hash(country)`) determines the target shuffle partition. - All rows with `country == 'India'` are routed to the exact same partition across the network. - Within each partition, duplicates are discarded. 2. **Execution Flow**: `select("country")` (Narrow) -> `distinct()` (Wide / Shuffle) -> `count()` (Action).

3. Step-by-Step Code / Logic Execution 1. Starting records: - Row 1: `{"country": "India"}` - Row 2: `{"country": "India"}` - Row 3: `{"country": "USA"}` 2. `df.select("country").distinct()`: - Duplicate `"India"` is eliminated. - Intermediate unique set: `[{"country": "India"}, {"country": "USA"}]`. 3. `.count()`: - Evaluates the total number of distinct elements: `2`.

4. Why the Correct Answer Is Optimal Option B is correct because: - The input contains 2 unique country values: 'India' and 'USA'. - `distinct()` correctly ensures uniqueness across the projected column.

5. Why Alternative Options Fail - Option A (1): Incorrect; "India" and "USA" are two distinct values, not one. - Option C (3): Incorrect; 3 represents the raw row count before deduplication. Failing to eliminate duplicates means `distinct()` was ignored. - Option D (4): Incorrect; `distinct()` never inflates row counts.

6. Edge Cases & Production Pitfalls - **NULL Values**: `distinct()` treats `NULL` as a distinct value. If the input were `["India", "India", None]`, `distinct().count()` would return `2` (one for "India" and one for `NULL`). - **High Cardinality Shuffles**: Running `distinct()` on a high-cardinality column (e.g., UUIDs) triggers a massive global shuffle across all cluster nodes, potentially spilling memory to disk.

7. Performance & Optimization Considerations - In PySpark, `df.distinct()` is an alias for `df.dropDuplicates()`. - For approximate counting on massive datasets where an exact number is unnecessary, use `approx_count_distinct("country", rsd=0.05)` to bypass the heavy shuffle.

8. Reliability & Operational Best Practices - Project only necessary columns before calling `distinct()`: `df.select("country").distinct()` shuffles only 1 string column, whereas `df.distinct()` on a 100-column table shuffles all 100 columns over the network.

9. Real-World Architectural Scenario A streaming media platform analyzes unique daily viewers across 180 countries. Running `df.select("user_id").distinct().count()` on 200 million daily log records caused shuffle timeouts. By migrating to HyperLogLog (`approx_count_distinct`), the analytics job completed in 45 seconds with 99% statistical accuracy.

10. Official Documentation & Learning Resources - Aggregations & Deduplication: [PySpark Aggregation & Distinct Operations Tutorial](/tutorials/pyspark/pyspark-aggregation)

1. Concept Overview The `filter()` transformation in PySpark evaluates a boolean column expression against each row in the DataFrame, retaining only those rows where the condition evaluates to `True`. The filter condition here is strictly greater than: `salary > 2000`. Evaluating the salaries [1000, 5000, 2000] reveals that only the row with salary `5000` satisfies the condition, resulting in an output count of `1`.

2. Architecture & Design Principle 1. **Narrow Transformation**: `filter()` is a **narrow dependency** transformation. Each executor can evaluate the boolean condition independently on its local in-memory partition partitions without network communication or data shuffling. 2. **Predicate Pushdown**: In production storage formats like Parquet and ORC, Spark pushes the filter predicate down to the storage layer (`PushedFilters: [GreaterThan(salary, 2000)]`), skipping entire row groups and files whose minimum and maximum statistics fall outside the threshold.

3. Step-by-Step Code / Logic Execution 1. Evaluate row 1: - ID: 1, Salary: 1000 - Condition: `1000 > 2000` -> `False` (Dropped) 2. Evaluate row 2: - ID: 2, Salary: 5000 - Condition: `5000 > 2000` -> `True` (Retained) 3. Evaluate row 3: - ID: 3, Salary: 2000 - Condition: `2000 > 2000` -> `False` (Strict inequality, dropped) 4. Aggregate count of retained rows: `1`.

4. Why the Correct Answer Is Optimal Option B is correct because: - The operator is strict inequality (`>`), not greater than or equal to (`>=`). Therefore, salary `2000` is excluded. - Exactly one record (ID 2, salary 5000) satisfies `salary > 2000`.

5. Why Alternative Options Fail - Option A (2): Incorrect; this would only be correct if the operator were `>=`, which would include salary 2000. - Option C (3): Incorrect; implies all rows matched the condition. - Option D (0): Incorrect; salary 5000 is strictly greater than 2000, so the filtered DataFrame is not empty.

6. Edge Cases & Production Pitfalls - **Three-Valued Logic with NULLs**: In PySpark, if `salary` contains a `NULL`, the condition `NULL > 2000` evaluates to `NULL` (unknown), and the row is excluded. - **Syntactic Aliasing**: `df.filter()` and `df.where()` are identical in PySpark; `where()` is provided for developers familiar with SQL syntax.

7. Performance & Optimization Considerations - Always filter data as early as possible in your transformation pipeline to reduce the volume of data transferred to subsequent stages ("Filter Early"). - Combine multiple filters with `&` (bitwise AND) rather than chained filters for Catalyst optimization clarity: `df.filter((col("salary") > 2000) & (col("dept") == "Eng"))`.

8. Reliability & Operational Best Practices - Wrap column conditions in parentheses when combining multiple expressions in PySpark (`(col("a") > 1) & (col("b") < 5)`), as Python operator precedence rules can cause syntax errors otherwise.

9. Real-World Architectural Scenario A financial fraud detection engine processes 50 million transactions per second. Applying strict predicate pushdown filters on transaction amount and merchant category at the storage read layer pruned 94% of unflagged transactions before join operations, reducing executor memory utilization by 12TB.

10. Official Documentation & Learning Resources - Data Filtering: [PySpark Filtering & Column Expressions Guide](/tutorials/pyspark/pyspark-filtering)

1. Concept Overview The `dropDuplicates()` method in PySpark removes duplicate rows from the DataFrame. When called without arguments, it evaluates all columns across the dataset and retains only the first occurrence of each unique combination of values. In the input DataFrame with rows [1, 1, 2], the duplicate row with `id=1` is dropped, leaving unique rows [1, 2], producing a final count of `2`.

2. Architecture & Design Principle 1. **Deduplication Mechanics**: Spark executes deduplication by assigning rows with identical column values to the same shuffle partition via hashing. 2. **Whole-Row vs. Subset Deduplication**: - `df.dropDuplicates()`: Considers every column in the schema. - `df.dropDuplicates(["col_a"])`: Considers only the specified subset of columns, dropping subsequent rows where `col_a` matches regardless of differences in other columns.

3. Step-by-Step Code / Logic Execution 1. Input records: - Row 1: `id = 1` - Row 2: `id = 1` - Row 3: `id = 2` 2. Execution of `dropDuplicates()`: - Row 1 is processed: unique set contains `{1}`. - Row 2 is evaluated: `id = 1` already exists in the unique set; row 2 is dropped. - Row 3 is processed: unique set contains `{1, 2}`. 3. `.count()` evaluates the size of the deduplicated DataFrame: `2`.

4. Why the Correct Answer Is Optimal Option A is correct because: - The input dataset has two duplicate records (`id = 1`) and one unique record (`id = 2`). - Eliminating one duplicate instance leaves exactly 2 records in the resulting DataFrame.

5. Why Alternative Options Fail - Option B (3): Incorrect; this is the row count before deduplication. - Option C (1): Incorrect; implies only a single record was retained. - Option D (Error): Incorrect; `dropDuplicates()` is a standard, fully supported PySpark DataFrame API method.

6. Edge Cases & Production Pitfalls - **Non-Deterministic Deduplication**: If multiple rows share the same key but differ in other columns (e.g., different timestamps), calling `df.dropDuplicates(["id"])` does not guarantee which row is preserved unless the data is explicitly sorted beforehand. To deterministically keep the latest record, use `row_number()` over a window ordered by timestamp descending. - **Shuffle Overhead**: Deduplicating across a 100-column DataFrame requires calculating hashes and shuffling all 100 columns across the network.

7. Performance & Optimization Considerations - Always pass a minimal list of business key columns to `dropDuplicates(subset=["key1", "key2"])` to avoid unnecessary data hashing and network serialization.

8. Reliability & Operational Best Practices - Combine `dropDuplicates()` with data validation checks: verify the row count delta (`before_count - after_count`) to detect abnormal duplicate spikes from upstream ingestion pipelines.

9. Real-World Architectural Scenario An IoT sensor pipeline receives 80 million telemetry pings per hour over MQTT. Due to cellular network reconnect retries, approximately 5% of sensor events arrive duplicated. Executing `df.dropDuplicates(subset=["sensor_id", "timestamp"])` cleans the stream before downstream aggregation tables are updated.

10. Official Documentation & Learning Resources - Dataframe Deduplication: [PySpark DataFrame Basics & Deduplication](/tutorials/pyspark/pyspark-dataframe-basics)

1. Concept Overview The `agg()` function in PySpark performs an aggregate calculation over the entire DataFrame (or grouped subsets). Passing `sum("amount")` instructs Spark to calculate the mathematical sum of all values in the `amount` column. Adding `100 + 200 + 300` equals `600`. Calling `.first()[0]` retrieves the first column value of the first row of the aggregated result, returning the numeric value `600`.

2. Architecture & Design Principle 1. **Two-Stage Aggregation**: - **Partial Aggregate (Map Phase)**: Each executor sums the values in its local partitions independently (`partial_sum`). - **Final Aggregate (Reduce Phase)**: Partial sums are shuffled and combined into a single total on the driver/reducer node. 2. **Execution Flow**: `agg(sum("amount"))` produces a single-row DataFrame `Row(sum(amount)=600)`. `.first()` pulls this single `Row` to the driver, and `[0]` accesses its first attribute.

3. Step-by-Step Code / Logic Execution 1. Input values in `amount` column: - Row 1: 100 - Row 2: 200 - Row 3: 300 2. Calculation: - Sum: `100 + 200 + 300 = 600`. 3. DataFrame evaluation: - `df.agg(sum("amount"))` -> DataFrame with schema `[sum(amount): bigint]` containing row `[600]`. - `.first()` -> `Row(sum(amount)=600)`. - `[0]` -> Extracts the first element: `600`.

4. Why the Correct Answer Is Optimal Option D is correct because: - The arithmetic sum of 100, 200, and 300 is 600. - `.first()[0]` extracts the computed scalar value directly from the aggregate row.

5. Why Alternative Options Fail - Option A (100): Incorrect; 100 is the first row's value, not the sum. - Option B (200): Incorrect; 200 is the arithmetic mean, not the sum. - Option C (300): Incorrect; 300 is the maximum individual value, not the total.

6. Edge Cases & Production Pitfalls - **NULL Values**: The `sum()` function automatically ignores `NULL` values. If a row had `amount = NULL`, the sum of `[100, 200, 300, NULL]` would still be `600`. If all rows are NULL, `sum()` returns `NULL`. - **Numeric Overflow**: Aggregating large integers can exceed 32-bit or 64-bit integer limits. In PySpark, `sum()` automatically upcasts integer types to `LongType` or `DoubleType` to prevent overflow.

7. Performance & Optimization Considerations - Multiple aggregations can be executed in a single pass: `df.agg(sum("amount"), avg("amount"), max("amount"))` avoids scanning the DataFrame multiple times.

8. Reliability & Operational Best Practices - Avoid calling `.first()[0]` on empty DataFrames without checking row counts first, as calling `.first()` on an empty dataset raises an `IndexError`.

9. Real-World Architectural Scenario An e-commerce checkout platform reconciles daily credit card settlement totals. Executing `df.agg(sum("net_amount")).first()[0]` across 50 million daily transaction rows verifies that the total captured revenue matches the bank gateway batch report down to the cent.

10. Official Documentation & Learning Resources - Aggregation Functions: [PySpark Aggregation & Statistical Operations](/tutorials/pyspark/pyspark-aggregation)

1. Concept Overview The `avg()` (or `mean()`) aggregation function in PySpark calculates the arithmetic mean of all numeric values in a column. The mean is computed as the sum of all values divided by the total count of non-null values: `(100 + 200 + 300) / 3 = 600 / 3 = 200`. Calling `.first()[0]` extracts the calculated scalar average value `200.0` (or `200`).

2. Architecture & Design Principle 1. **Distributed Average Computation**: - Because data is distributed across multiple partitions, Spark does not compute the average by dividing locally on each partition. - Instead, each executor tracks two accumulators for its partition: `sum` and `count`. - The final stage sums all partial sums and all partial counts, performing the division `total_sum / total_count` only in the final aggregation step.

3. Step-by-Step Code / Logic Execution 1. Extract column values: `[100, 200, 300]`. 2. Compute sum: `100 + 200 + 300 = 600`. 3. Compute count of non-null entries: `3`. 4. Divide sum by count: `600 / 3 = 200`. 5. `df.agg(avg("amount")).first()[0]` retrieves the computed scalar: `200`.

4. Why the Correct Answer Is Optimal Option D is correct because: - The arithmetic average of 100, 200, and 300 is exactly 200. - `avg()` properly computes the mean across all records.

5. Why Alternative Options Fail - Option A (100): Incorrect; 100 is the minimum value. - Option B (600): Incorrect; 600 is the total sum, not the average. - Option C (300): Incorrect; 300 is the maximum value.

6. Edge Cases & Production Pitfalls - **NULL Handling**: `avg()` ignores NULL values in both the numerator (sum) and denominator (count). For example, the average of `[100, 200, NULL]` is `(100 + 200) / 2 = 150`, not `100`. - **Zero Rows**: If the DataFrame is completely empty or all values are NULL, `avg()` returns `NULL` (`None` in Python).

7. Performance & Optimization Considerations - In PySpark, `avg()` returns a `DoubleType` by default. If exact decimal precision is required for financial applications, cast the column to `DecimalType(18, 4)` before aggregating.

8. Reliability & Operational Best Practices - When analyzing skewed datasets with extreme outliers, complement `avg()` with median and percentile calculations (`percentile_approx()`) to prevent outliers from distorting metrics.

9. Real-World Architectural Scenario A rideshare logistics platform tracks driver trip durations. Running `df.agg(avg("duration_mins"))` across 10 million daily rides provides the baseline average trip time for urban traffic optimization models.

10. Official Documentation & Learning Resources - Aggregation & Math: [PySpark Aggregations & GroupBy Guide](/tutorials/pyspark/pyspark-aggregation)

1. Concept Overview The `limit(n)` transformation in PySpark restricts the DataFrame to at most the first `n` rows. Passing `2` to `df.limit(2)` truncates the dataset from 3 rows down to 2 rows. Calling the action `count()` on the resulting DataFrame counts those remaining rows and outputs `2`.

2. Architecture & Design Principle 1. **Local vs. Global Limit**: - In Spark's Catalyst optimizer, `limit(n)` is executed as a two-stage physical plan: - **LocalLimit(n)**: Each partition yields at most `n` rows locally. - If the first partition already contains `n` rows, Spark can short-circuit and avoid scanning remaining partitions. - **GlobalLimit(n)**: Gathers the results to a single partition and takes the first `n` rows globally. 2. **Early Termination**: Unlike full table scans, `limit()` enables Spark tasks to terminate early once the requested row quota is met.

3. Step-by-Step Code / Logic Execution 1. Original DataFrame contains 3 rows: `[1, 2, 3]`. 2. `df.limit(2)` executes: - Spark fetches the first 2 rows encountered. - The resulting DataFrame contains exactly 2 rows. 3. `.count()` evaluates the row count of this limited dataset: `2`.

4. Why the Correct Answer Is Optimal Option D is correct because: - `limit(2)` explicitly constrains the row count to 2. - The downstream action `count()` returns the exact number of rows present in the limited DataFrame.

5. Why Alternative Options Fail - Option A (1): Incorrect; the limit specified was 2, not 1. - Option B (3): Incorrect; ignores the effect of `limit(2)`. - Option C (Error): Incorrect; `limit()` is a valid core DataFrame API method.

6. Edge Cases & Production Pitfalls - **Non-Deterministic Order**: Without an explicit `orderBy()`, `df.limit(2)` returns an arbitrary set of 2 rows depending on partition scan order across executors. To guarantee consistent rows across runs, always sort first: `df.orderBy("id").limit(2)`. - **Limit Higher Than Dataset Size**: If you specify `df.limit(100)` on a dataset with only 3 rows, Spark returns all 3 rows without error.

7. Performance & Optimization Considerations - Use `limit()` when inspecting sample data in development notebooks to prevent driver out-of-memory errors.

8. Reliability & Operational Best Practices - Never assume `limit(n)` returns rows in insertion order; distributed file systems (S3/HDFS) do not preserve physical row ordering unless an explicit sort key is defined.

9. Real-World Architectural Scenario A customer intelligence pipeline generates a daily executive email listing the Top 10 highest-spending accounts. The query sorts accounts by revenue descending and applies `.limit(10)`, allowing Spark to prune millions of remaining customer records and return the top records in seconds.

10. Official Documentation & Learning Resources - Sorting and Limits: [PySpark Sorting Sampling & Limit Operations](/tutorials/pyspark/sorting-sampling-limit-pyspark)

1. Concept Overview The `fillna()` (or `na.fill()`) transformation in PySpark replaces null (`None`) values in a DataFrame with specified replacement values. In this code, passing the numeric integer `0` to `df.fillna(0)` instructs Spark to identify all numeric columns containing `NULL` and replace those null entries with `0`. Thus, the `NULL` in the first row becomes `0`.

2. Architecture & Design Principle 1. **Type-Safe Replacement**: Spark's `fillna()` matches replacement types to column schemas: - Passing an integer (e.g., `0`) automatically replaces nulls in integer, long, float, and double columns. - Passing a string (e.g., `"Unknown"`) replaces nulls only in string columns. 2. **Narrow Transformation**: `fillna()` is a column-level transformation evaluated partition by partition without any shuffle across the network.

3. Step-by-Step Code / Logic Execution 1. Input DataFrame schema: `[value: integer]` with rows: - Row 1: `NULL` - Row 2: `100` 2. When `df.fillna(0)` executes: - Row 1: `value` is `NULL`; replaces `NULL` with `0`. - Row 2: `value` is `100`; non-null, remains `100`. 3. The resulting DataFrame has rows: `[0, 100]`. The original `NULL` becomes `0`.

4. Why the Correct Answer Is Optimal Option C is correct because: - `fillna(0)` replaces null values with the specified argument `0`. - The transformation produces a clean, null-free numeric column suitable for downstream arithmetic operations.

5. Why Alternative Options Fail - Option A (NULL): Incorrect; `fillna()` explicitly replaces NULL values. - Option B (Empty String): Incorrect; the column is numeric, and the fill argument passed is the integer `0`, not `""`. - Option D (100): Incorrect; 100 is the value of the second row, not the replacement for NULL.

6. Edge Cases & Production Pitfalls - **Targeting Specific Columns**: Passing a scalar (`df.fillna(0)`) modifies *all* compatible numeric columns. To target specific columns, pass a dictionary: `df.fillna({"value": 0, "status": "PENDING"})` or specify the subset: `df.fillna(0, subset=["value"])`. - **NaN vs. NULL**: In Spark, IEEE `NaN` (Not-a-Number) in floating point columns is technically distinct from SQL `NULL`. To replace both NaNs and NULLs, ensure appropriate cleaning functions are applied.

7. Performance & Optimization Considerations - `fillna()` introduces minimal Catalyst overhead and evaluates during partition scans with zero data shuffling.

8. Reliability & Operational Best Practices - Always clean and impute null values before training machine learning models (Spark MLlib), as estimators like LinearRegression and VectorAssembler throw exceptions when encountering unhandled nulls.

9. Real-World Architectural Scenario In a retail point-of-sale reporting pipeline, item discount columns frequently contained NULL when no promo code was applied. Downstream revenue calculations (`price - discount`) produced NULL outputs whenever discount was NULL. Applying `df.fillna(0, subset=["discount"])` ensured all sales totals computed accurately.

10. Official Documentation & Learning Resources - Handling Missing Values: [PySpark Missing & Null Values Guide](/tutorials/pyspark/pyspark-missing)

1. Concept Overview An **inner join** in PySpark combines rows from two DataFrames where the join key matches in both datasets. Any row from either side that does not have a corresponding match on the specified join condition is excluded from the output. In this scenario, `sales_df` has IDs [1, 2] and `customer_df` has IDs [2, 3]. The only ID present in both tables is `2`, yielding exactly `1` joined output row.

2. Architecture & Design Principle 1. **Join Execution Strategies**: - **Sort Merge Join (SMJ)**: Standard for large datasets. Both DataFrames are shuffled on the join key (`id`), sorted by key within partitions, and merged. - **Broadcast Hash Join (BHJ)**: If one table is small (<10MB default threshold), Spark broadcasts the small table to all executor memory nodes, avoiding expensive network shuffles of the large table. 2. **Inner Join Logic**: Outputs the intersection of keys: `Keys(sales) ∩ Keys(customer) = {1, 2} ∩ {2, 3} = {2}`.

3. Step-by-Step Code / Logic Execution 1. Evaluate left table (`sales_df`): IDs `{1, 2}`. 2. Evaluate right table (`customer_df`): IDs `{2, 3}`. 3. Inner join condition `sales_df.id == customer_df.id`: - ID 1: Exists in sales, missing in customer -> Not matched, omitted. - ID 2: Exists in sales, exists in customer -> Matched! Output row produced. - ID 3: Missing in sales, exists in customer -> Not matched, omitted. 4. Total resulting rows = `1`.

4. Why the Correct Answer Is Optimal Option C is correct because: - Only ID 2 satisfies the inner join condition by appearing in both DataFrames. - The resulting DataFrame has exactly 1 row.

5. Why Alternative Options Fail - Option A (4): Incorrect; represents the Cartesian product (cross join) of 2 x 2 rows. - Option B (2): Incorrect; represents the row count of an outer or left join. - Option D (0): Incorrect; ID 2 matches, so the join produces output.

6. Edge Cases & Production Pitfalls - **Duplicate Keys**: If `sales_df` had two rows with `id = 2` and `customer_df` had two rows with `id = 2`, an inner join would produce `2 x 2 = 4` rows. Always verify key uniqueness before joining. - **Null Join Keys**: In SQL and PySpark standard joins, `NULL == NULL` evaluates to false. Rows with `NULL` as the join key never match in an inner join.

7. Performance & Optimization Considerations - When joining a large fact table with a small dimension table, use explicit broadcast hinting: `sales_df.join(broadcast(customer_df), on="id")` to force a Broadcast Hash Join and bypass shuffle stages.

8. Reliability & Operational Best Practices - Specify `on="id"` as a string or list of strings when the join column name is identical in both DataFrames. This retains a single `id` column in the output, avoiding ambiguous duplicate column references downstream.

9. Real-World Architectural Scenario A telecom billing system joins daily call records with customer billing plans. Joining a 10-billion-row call detail table with an un-deduplicated account dimension produced massive row multiplication, causing executor memory crashes. Enforcing strict primary key uniqueness before joining resolved the data skew and halved cluster runtime.

10. Official Documentation & Learning Resources - DataFrame Joins: [PySpark Joins & Join Optimization Tutorial](/tutorials/pyspark/df-joins)

1. Concept Overview A **left outer join** (or simply left join) preserves all rows from the left DataFrame (`sales_df`), regardless of whether a matching row exists in the right DataFrame (`customer_df`). If a match is found on the right, the right columns are populated; if no match exists, the right columns are filled with `NULL`. Because `sales_df` contains 2 rows and each row matches at most one record in `customer_df`, the resulting DataFrame contains exactly `2` rows.

2. Architecture & Design Principle 1. **Preservation of Left Domain**: In relational algebra, a Left Outer Join guarantees that every record from the left relation appears in the output at least once. 2. **Distributed Shuffle Mechanics**: - Spark shuffles rows from both tables using hash partitioning on `id`. - On each partition, Spark scans left rows: - For `id = 1`: No right match -> emits row with `id = 1` and null right attributes. - For `id = 2`: Right match found -> emits joined row. 3. Row count is strictly determined by the cardinality of the left table when right keys are unique.

3. Step-by-Step Code / Logic Execution 1. Left table (`sales_df`): IDs `[1, 2]` (2 rows). 2. Right table (`customer_df`): IDs `[2, 3]` (2 rows). 3. Left Join evaluation: - ID 1 from `sales_df`: Evaluated against right table. No match for ID 1. Emits `[id: 1, customer_cols: NULL]`. - ID 2 from `sales_df`: Evaluated against right table. Match found for ID 2. Emits `[id: 2, customer_cols: matched_data]`. 4. Total resulting rows: `2`.

4. Why the Correct Answer Is Optimal Option A is correct because: - All 2 records from the left DataFrame (`sales_df`) are preserved. - The right table has no duplicate keys for ID 2, meaning no row expansion occurs.

5. Why Alternative Options Fail - Option B (1): Incorrect; 1 represents the inner join count, omitting ID 1. - Option C (3): Incorrect; 3 would be the row count of a full outer join (`{1, 2, 3}`). - Option D (4): Incorrect; represents Cartesian product expansion.

6. Edge Cases & Production Pitfalls - **Right Key Duplication**: If the right DataFrame had three records with `id = 2`, the left row with `id = 2` would match all three, expanding the total row count to `1 (for id 1) + 3 (for id 2) = 4`. A left join only guarantees *at least* the left row count, not *at most*. - **Post-Join Filters**: Placing a filter like `.filter(col("right_col") == "active")` on a left join inadvertently converts it into an inner join because rows with `NULL` right values are dropped! Use `.filter((col("right_col") == "active") | col("right_col").isNull())` if nulls must be preserved.

7. Performance & Optimization Considerations - If `customer_df` is small, broadcast it with `broadcast(customer_df)` to eliminate shuffling `sales_df`.

8. Reliability & Operational Best Practices - Use left anti joins (`how="left_anti"`) if your goal is solely to find records in the left table that do *not* exist in the right table.

9. Real-World Architectural Scenario A banking warehouse joins loan applications (left table) with fraud blacklists (right table). A left join ensures every loan application is preserved in the pipeline, appending fraud flags where matches occur while keeping clean loans intact.

10. Official Documentation & Learning Resources - Join Types in PySpark: [PySpark Joins & Execution Plans](/tutorials/pyspark/df-joins)

1. Concept Overview The `rank()` window function assigns a rank to each row within an ordered window partition. When duplicate (tied) values are encountered in the ordering column, `rank()` assigns the **same rank** to all tied rows, but **skips ranks** for subsequent rows corresponding to the number of ties. For salaries `[1000, 1000, 2000]`, the two tied salaries at 1000 receive rank `1`, and the next salary skips rank 2 to receive rank `3`, producing the sequence `1, 1, 3`.

2. Architecture & Design Principle 1. **Window Specification Mechanics**: - `Window.orderBy("salary")` establishes the logical frame across which the ranking evaluation occurs. - Without a `partitionBy()`, all rows are routed to a single partition to determine global ordering. 2. **Standard Competition Ranking ("1224" Rule)**: - Rank reflects the number of preceding rows with strictly smaller values plus one. - For row 1 (1000): 0 smaller rows -> Rank = `0 + 1 = 1`. - For row 2 (1000): 0 smaller rows -> Rank = `0 + 1 = 1` (tie). - For row 3 (2000): 2 smaller rows -> Rank = `2 + 1 = 3` (Rank 2 is skipped).

3. Step-by-Step Code / Logic Execution 1. Sort dataset by `salary` ascending: `[1000, 1000, 2000]`. 2. Evaluate row 1 (1000): - First position: Rank = 1. 3. Evaluate row 2 (1000): - Matches salary of row 1 (tie): Rank = 1. 4. Evaluate row 3 (2000): - Two records precede this value: Rank = 3. 5. Final column output: `[1, 1, 3]`.

4. Why the Correct Answer Is Optimal Option B is correct because: - `rank()` assigns identical ranks to tied values and skips numbers after ties to reflect physical row position. - Ranks 1, 1, 3 accurately represent standard competition ranking.

5. Why Alternative Options Fail - Option A (1, 1, 2): Incorrect; this is the behavior of `dense_rank()`, which does not skip ranks after ties. - Option C (1, 2, 3): Incorrect; this is the behavior of `row_number()`, which assigns strictly sequential integers regardless of ties. - Option D (2, 2, 3): Incorrect; ranks are 1-indexed.

6. Edge Cases & Production Pitfalls - **Global Ordering Without PartitionBy**: Defining a window with `orderBy` but omitting `partitionBy` forces all data into a single Spark partition on one executor core, creating a severe cluster bottleneck on large datasets. - **NULL Ordering**: By default in Spark, `NULLS FIRST` applies for ascending sorts, giving nulls the lowest ranks unless `.asc_nulls_last()` is specified.

7. Performance & Optimization Considerations - Always include `partitionBy("department")` or an equivalent grouping key when working with large tables to distribute window evaluations across the entire cluster.

8. Reliability & Operational Best Practices - Choose the correct ranking function based on business specifications: - Use `row_number()` when each row must have a unique sequential ID. - Use `rank()` for competition leaderboards where ties push subsequent ranks down. - Use `dense_rank()` for consecutive tiering without gaps (e.g., top 3 salary tiers).

9. Real-World Architectural Scenario In a digital gaming leaderboard, two players tied for 1st place with 10,000 points. The next player with 9,500 points should receive the 3rd-place bronze medal, not silver. Using `rank()` accurately skips 2nd place and assigns bronze to 3rd.

10. Official Documentation & Learning Resources - Window Functions: [PySpark DataFrame Window Functions Deep Dive](/tutorials/pyspark/df-window-functions)

1. Concept Overview The `dense_rank()` window function assigns ranks to rows within an ordered window partition such that tied values receive the **same rank**, but **no ranks are skipped** after ties. Unlike `rank()`, which leaves gaps, `dense_rank()` produces a continuous, gapless sequence of integer ranks. For salaries `[1000, 1000, 2000]`, the two tied salaries at 1000 receive rank `1`, and the next salary of 2000 receives rank `2`, yielding the sequence `1, 1, 2`.

2. Architecture & Design Principle 1. **Dense Ranking Mechanics**: - Ranks represent distinct value levels rather than row counts. - Each unique value in the sorted column increments the rank by exactly 1. - Unique values: `Distinct(1000, 2000) -> [1000: Rank 1, 2000: Rank 2]`. 2. **Memory Evaluation**: Spark evaluates dense ranks within each window frame by retaining the previous row's sort value and incrementing the rank counter only when the sort value changes.

3. Step-by-Step Code / Logic Execution 1. Sort rows by `salary` ascending: `[1000, 1000, 2000]`. 2. Evaluate row 1 (1000): - Distinct value level 1 -> `dense_rank = 1`. 3. Evaluate row 2 (1000): - Same value as row 1 -> `dense_rank = 1`. 4. Evaluate row 3 (2000): - Value changed; distinct value level 2 -> `dense_rank = 2`. 5. Output sequence: `[1, 1, 2]`.

4. Why the Correct Answer Is Optimal Option A is correct because: - `dense_rank()` guarantees no gaps in ranking sequences. - 1000 is tier 1, and 2000 is tier 2.

5. Why Alternative Options Fail - Option B (1, 1, 3): Incorrect; this is `rank()`, which skips rank 2. - Option C (1, 2, 3): Incorrect; this is `row_number()`, which assigns unique sequential integers regardless of value equivalence. - Option D (2, 2, 3): Incorrect; dense ranking begins at index 1.

6. Edge Cases & Production Pitfalls - **"Top N" Queries**: When asking for the "Top 3 highest salaries", using `dense_rank() <= 3` returns all employees earning within the top 3 salary tiers (which may be more than 3 employees if ties exist). If strictly 3 employees are required, use `row_number() <= 3`.

7. Performance & Optimization Considerations - Always supply `partitionBy()` whenever possible to parallelize window execution across worker nodes.

8. Reliability & Operational Best Practices - Clearly distinguish business requirements between "Top N Entities" (requires `row_number`) and "Top N Values/Tiers" (requires `dense_rank`).

9. Real-World Architectural Scenario An HR analytics dashboard computes executive compensation tiers across 20 departments. Because multiple vice presidents earn identical base salaries, the business requires a tiering system without gaps. `dense_rank()` groups all tied executives into Tier 1 and correctly designates the next compensation bracket as Tier 2.

10. Official Documentation & Learning Resources - Window Functions: [PySpark Window Functions & Dense Rank Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview The `lag(col, offset)` window function in PySpark accesses data from a previous row at a specified physical offset within the window frame. Specifying an offset of `1` (`lag("salary", 1)`) retrieves the value of `salary` from the immediately preceding row. For the second row (where salary = 2000), the immediately preceding row has salary `1000`, so `prev` evaluates to `1000`.

2. Architecture & Design Principle 1. **Window Frame Traversal**: - Spark processes rows sequentially within each sorted partition. - For `lag(col, 1)`, Spark maintains a buffer of previous row values in memory. - When evaluating row `i`, it reads the value from row `i - 1`. 2. **Boundary Handling**: For the very first row (`i = 0`), no preceding row exists within the partition, so `lag()` returns `NULL` (or an optional user-defined default value).

3. Step-by-Step Code / Logic Execution 1. Sorted order of `salary`: `[1000, 2000, 3000]`. 2. Row 1 (salary = 1000): - Preceding row: None exists. - `prev` = `NULL`. 3. Row 2 (salary = 2000): - Preceding row has salary = `1000`. - `prev` = `1000`. 4. Row 3 (salary = 3000): - Preceding row has salary = `2000`. - `prev` = `2000`. 5. Therefore, for row 2, the value is `1000`.

4. Why the Correct Answer Is Optimal Option C is correct because: - The previous row relative to salary 2000 is 1000. - Offset 1 correctly steps back by one row.

5. Why Alternative Options Fail - Option A (NULL): Incorrect; NULL is returned for the first row (salary 1000), not the second. - Option B (3000): Incorrect; 3000 is the subsequent row, which would be retrieved by `lead()`, not `lag()`. - Option D (2000): Incorrect; 2000 is the current row's value.

6. Edge Cases & Production Pitfalls - **Default Value Parameter**: `lag()` accepts a third argument: `lag("salary", 1, 0)`. If provided, the first row evaluates to `0` instead of `NULL`, preventing null pointer exceptions in arithmetic operations. - **Partition Boundaries**: `lag()` never crosses partition boundaries defined by `partitionBy()`. The first row of each partition always evaluates to `NULL`.

7. Performance & Optimization Considerations - `lag()` requires keeping previous rows in an in-memory buffer. Avoid excessively large offsets (e.g., `lag("col", 10000)`) as this increases executor memory consumption.

8. Reliability & Operational Best Practices - Always ensure `orderBy()` is explicitly specified when using `lag()`; without deterministic ordering, the "previous" row is unpredictable across cluster runs.

9. Real-World Architectural Scenario A financial market pipeline computes daily stock price fluctuations. Running `df.withColumn("price_diff", col("close_price") - lag("close_price", 1).over(Window.partitionBy("ticker").orderBy("date")))` accurately calculates daily price changes for millions of equities.

10. Official Documentation & Learning Resources - Window Functions: [PySpark DataFrame Window Functions Tutorial](/tutorials/pyspark/df-window-functions)

1. Concept Overview The `lead(col, offset)` window function in PySpark looks forward to retrieve data from a subsequent row at a specified physical offset within the window frame. Specifying an offset of `1` (`lead("salary", 1)`) retrieves the value of `salary` from the immediately following row. For the second row (where salary = 2000), the immediately following row has salary `3000`, so `next` evaluates to `3000`.

2. Architecture & Design Principle 1. **Forward Row Access**: - In distributed stream and batch processing, `lead()` allows looking ahead without requiring self-joins. - When evaluating row `i`, Spark accesses row `i + 1` from the partition buffer. 2. **Terminal Boundary Condition**: For the final row in an ordered partition, no subsequent row exists; `lead()` returns `NULL` (or an optional default value).

3. Step-by-Step Code / Logic Execution 1. Sorted sequence of salaries: `[1000, 2000, 3000]`. 2. Row 1 (salary = 1000): - Following row has salary = 2000. - `next` = `2000`. 3. Row 2 (salary = 2000): - Following row has salary = 3000. - `next` = `3000`. 4. Row 3 (salary = 3000): - Following row: None exists (end of partition). - `next` = `NULL`. 5. Therefore, for the second row, the value is `3000`.

4. Why the Correct Answer Is Optimal Option C is correct because: - The next row following salary 2000 is 3000. - `lead()` steps forward by exactly 1 position.

5. Why Alternative Options Fail - Option A (1000): Incorrect; 1000 is the preceding row retrieved by `lag()`. - Option B (NULL): Incorrect; NULL is returned for the last row (salary 3000), not the middle row. - Option D (2000): Incorrect; 2000 is the current row's value.

6. Edge Cases & Production Pitfalls - **Default Replacement**: Like `lag()`, `lead()` takes an optional third argument for the default value when an offset points beyond partition boundaries: `lead("salary", 1, 0)`. - **Partition Spanning**: `lead()` never looks across different partitions; the final row of every partition evaluates to `NULL`.

7. Performance & Optimization Considerations - Always supply `partitionBy()` to parallelize evaluation across executor tasks.

8. Reliability & Operational Best Practices - Never use self-joins when `lag()` or `lead()` can accomplish the same result. A window function processes data in a single shuffle pass, whereas a self-join requires two separate scans and an expensive join shuffle.

9. Real-World Architectural Scenario An IoT logistics system tracks vehicle GPS pings. To determine how long a delivery van remained at a delivery stop, the pipeline calculates `lead("event_time", 1).over(w) - col("event_time")`, finding the timestamp of the next movement without complex self-joins.

10. Official Documentation & Learning Resources - Window Operations: [PySpark DataFrame Window Functions Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview The `row_number()` window function assigns a unique, strictly sequential integer starting from 1 to each row within an ordered window partition. Because there are three rows ('A', 'B', 'C') ordered ascending, `row_number()` numbers them sequentially: row 1 receives `1`, row 2 receives `2`, and row 3 receives `3`, producing the sequence `1, 2, 3`.

2. Architecture & Design Principle 1. **Strict Monotonicity**: - Unlike `rank()` or `dense_rank()`, `row_number()` never produces ties or skips numbers. - Even if two rows have identical values in the `orderBy` column, `row_number()` deterministically assigns distinct numbers (e.g., 1 and 2). 2. **1-Based Indexing**: SQL and Spark window functions follow standard 1-based indexing conventions, beginning at `1` rather than `0`.

3. Step-by-Step Code / Logic Execution 1. Sort column `value` alphabetically ascending: `['A', 'B', 'C']`. 2. Row 1 ('A'): Assigned `row_number = 1`. 3. Row 2 ('B'): Assigned `row_number = 2`. 4. Row 3 ('C'): Assigned `row_number = 3`. 5. Output sequence: `[1, 2, 3]`.

4. Why the Correct Answer Is Optimal Option C is correct because: - `row_number()` starts at 1 and increments by 1 for each successive record. - The output contains exactly three rows numbered 1, 2, 3.

5. Why Alternative Options Fail - Option A (3, 2, 1): Incorrect; would require descending order (`orderBy(col("value").desc())`). - Option B (0, 1, 2): Incorrect; Spark window numbering is 1-based, not 0-based. - Option D (1, 1, 1): Incorrect; `row_number()` never assigns identical numbers to different rows.

6. Edge Cases & Production Pitfalls - **Non-Deterministic Ties**: If rows have identical values in the sort column, Spark breaks ties arbitrarily unless secondary sort columns are specified (e.g., `Window.orderBy("value", "id")`). - **Deduplication Pattern**: A common production pattern for deduplication is `df.withColumn("rn", row_number().over(Window.partitionBy("id").orderBy(col("updated_at").desc()))).filter(col("rn") == 1)`.

7. Performance & Optimization Considerations - Global `row_number()` over an entire table without `partitionBy()` moves all data to a single partition, destroying cluster parallelism.

8. Reliability & Operational Best Practices - Always specify a deterministic tie-breaker in `orderBy()` if consistent row numbering across multiple job runs is required.

9. Real-World Architectural Scenario In a Change Data Capture (CDC) pipeline, updates to customer addresses arrive as separate records with timestamps. To reconstruct the current state of each customer, the pipeline partitions by `customer_id`, orders by `update_timestamp.desc()`, and filters for `row_number() == 1`.

10. Official Documentation & Learning Resources - Window Numbering: [PySpark Window Functions & Row Number Tutorial](/tutorials/pyspark/df-window-functions)

1. Concept Overview A **running total** (cumulative sum) calculates the progressive sum of a column from the start of the window frame up to the current row. The frame specification `rowsBetween(Window.unboundedPreceding, Window.currentRow)` defines the aggregation window as all rows from the beginning of the partition up to and including the current row. For amounts `[100, 200, 300]`, the cumulative sums are `100`, `100 + 200 = 300`, and `100 + 200 + 300 = 600`, producing the sequence `100, 300, 600`.

2. Architecture & Design Principle 1. **Window Framing (ROWS vs. RANGE)**: - **ROWS**: Considers physical row offsets. `rowsBetween(Window.unboundedPreceding, Window.currentRow)` aggregates strictly from row 0 to the current row index. - **Default Frame**: When `orderBy()` is specified without an explicit frame, SQL and Spark default to `rangeBetween(Window.unboundedPreceding, Window.currentRow)`. For unique values, this yields identical results; but for tied values, `RANGE` sums all tied rows together. Explicitly using `rowsBetween` prevents unexpected tie coalescing.

3. Step-by-Step Code / Logic Execution 1. Sorted input rows: `[100, 200, 300]`. 2. Row 1 (100): - Frame includes: `[100]`. - Running total = `100`. 3. Row 2 (200): - Frame includes: `[100, 200]`. - Running total = `100 + 200 = 300`. 4. Row 3 (300): - Frame includes: `[100, 200, 300]`. - Running total = `100 + 200 + 300 = 600`. 5. Output sequence: `[100, 300, 600]`.

4. Why the Correct Answer Is Optimal Option B is correct because: - The cumulative summation of 100, then +200, then +300 correctly yields 100, 300, 600. - The window frame accumulates accurately on each successive row.

5. Why Alternative Options Fail - Option A (100, 200, 300): Incorrect; returns the original input column values without accumulation. - Option C (600, 600, 600): Incorrect; represents an unconstrained window over the entire partition (`rowsBetween(unboundedPreceding, unboundedFollowing)`). - Option D (100, 100, 100): Incorrect; does not accumulate subsequent rows.

6. Edge Cases & Production Pitfalls - **Tied Values with Default RANGE**: If two rows have identical amounts (e.g., `[100, 100]`), default `RANGE` framing produces `[200, 200]` because ties are treated as belonging to the same range frame. Using explicit `rowsBetween` ensures progressive `[100, 200]`.

7. Performance & Optimization Considerations - When computing running totals across large tables, always partition by customer, account, or category (`partitionBy("customer_id")`) so Spark evaluates running totals in parallel across partitions.

8. Reliability & Operational Best Practices - Explicitly declare the window frame (`rowsBetween` or `rangeBetween`) in production code to prevent behavioral surprises caused by default frame definitions.

9. Real-World Architectural Scenario A banking ledger pipeline displays running balances on checking accounts. Each deposit or withdrawal updates the cumulative balance using `sum("transaction_amount").over(Window.partitionBy("account_id").orderBy("txn_timestamp").rowsBetween(Window.unboundedPreceding, Window.currentRow))`.

10. Official Documentation & Learning Resources - Cumulative Aggregations: [PySpark Window Framing & Cumulative Sums Guide](/tutorials/pyspark/df-window-functions)

1. Concept Overview In PySpark, `df.union(other_df)` is equivalent to the SQL **UNION ALL** operation—it concatenates the rows of two DataFrames without deduplicating identical records. Combining a 2-row DataFrame with itself using `union()` results in `2 + 2 = 4` rows. Calling `count()` returns `4`.

2. Architecture & Design Principle 1. **Narrow Dependency**: - Because `union()` does not deduplicate, it requires **no data shuffle**. - Partitions from the first DataFrame and partitions from the second DataFrame are combined directly into the new DataFrame's partition list. 2. **SQL Standard Divergence**: - In standard ANSI SQL, `UNION` implies `UNION DISTINCT` (deduplicating results), whereas `UNION ALL` preserves duplicates. - In PySpark, `df.union()` performs `UNION ALL` for performance reasons. To achieve SQL `UNION DISTINCT`, you must explicitly chain `df.union(other_df).distinct()`.

3. Step-by-Step Code / Logic Execution 1. Original DataFrame `df` contains 2 rows: `[1, 2]`. 2. Execution of `df.union(df)`: - Left branch yields: `[1, 2]`. - Right branch yields: `[1, 2]`. - Combined DataFrame contains: `[1, 2, 1, 2]`. 3. `.count()` evaluates the total number of rows: `4`.

4. Why the Correct Answer Is Optimal Option D is correct because: - PySpark's `union()` preserves duplicate records. - Concatenating two 2-row DataFrames produces exactly 4 rows.

5. Why Alternative Options Fail - Option B (2): Incorrect; 2 would only be the result if `distinct()` were called after `union()`. - Option A (1) & Option C (3): Incorrect; mathematical concatenation of 2 + 2 equals 4.

6. Edge Cases & Production Pitfalls - **Column Resolution by Position vs. Name**: - `df.union()` resolves columns **by position**, not by column name! If `df1` has columns `[age, name]` and `df2` has `[name, age]`, `df1.union(df2)` places strings into the integer column and integers into the string column. - To union safely by column name, always use `df1.unionByName(df2)`. - **Missing Columns with unionByName**: In Spark 3.1+, `df1.unionByName(df2, allowMissingColumns=True)` automatically fills missing columns in either table with `NULL`.

7. Performance & Optimization Considerations - Chaining dozens of `union()` calls in a loop creates an extremely deep Catalyst query plan, leading to `StackOverflowError` during optimization. Use `spark.createDataFrame()` or batch unions with `reduce()` instead.

8. Reliability & Operational Best Practices - In enterprise pipelines, standardize on `unionByName()` to prevent silent data corruption caused by schema column reordering.

9. Real-World Architectural Scenario A multi-region e-commerce platform loads daily transaction files from US, EU, and APAC object stores. Using `us_df.unionByName(eu_df).unionByName(apac_df)` consolidates regional sales into a unified global DataFrame with zero shuffle overhead.

10. Official Documentation & Learning Resources - DataFrame Transformations: [PySpark DataFrame Basics & Combining Datasets](/tutorials/pyspark/pyspark-dataframe-basics)

1. Concept Overview The `filter()` transformation applies a boolean condition to each row of the DataFrame. The condition `col("id") > 1` tests whether the value of `id` is strictly greater than 1. For rows with IDs 1, 2, and 3: - ID 1: `1 > 1` is False (dropped). - ID 2: `2 > 1` is True (retained). - ID 3: `3 > 1` is True (retained). The resulting DataFrame contains the IDs `2, 3`.

2. Architecture & Design Principle 1. **Row-by-Row Narrow Evaluation**: - Each partition evaluates the boolean expression in parallel. - Memory records failing the predicate are discarded without ever being transferred over the cluster network. 2. **Column Projection**: `select("id")` narrows the output schema, which in this single-column table retains `id`.

3. Step-by-Step Code / Logic Execution 1. Input records: `[1, 2, 3]`. 2. Row 1: `id = 1`. Condition `1 > 1` -> False. 3. Row 2: `id = 2`. Condition `2 > 1` -> True. 4. Row 3: `id = 3`. Condition `3 > 1` -> True. 5. Filtered result: `[2, 3]`.

4. Why the Correct Answer Is Optimal Option D is correct because: - The operator is strict inequality (`>`), excluding 1. - Both 2 and 3 are strictly greater than 1.

5. Why Alternative Options Fail - Option A (1): Incorrect; 1 is the only row excluded by the filter. - Option B (3): Incorrect; omits ID 2, which also satisfies `id > 1`. - Option C (1, 2): Incorrect; includes 1 and omits 3.

6. Edge Cases & Production Pitfalls - **Type Coercion**: If `id` were stored as a string column (`StringType`), string comparison rules apply (`"10" < "2"`). Ensure numeric columns are cast to `IntegerType` or `LongType` before applying numerical comparisons.

7. Performance & Optimization Considerations - In Parquet and Delta Lake, min/max file statistics allow Spark to skip reading entire files if the file's maximum `id` is <= 1.

8. Reliability & Operational Best Practices - Always test filter conditions on boundary values (e.g., testing with values 0, 1, and 2) to ensure strict inequalities (`>`) versus non-strict inequalities (`>=`) match business requirements.

9. Real-World Architectural Scenario An authentication pipeline processes active user login sessions, filtering out unverified accounts with `df.filter(col("verification_level") > 1)` before issuing access tokens.

10. Official Documentation & Learning Resources - Filtering Techniques: [PySpark Filtering & Column Expressions Guide](/tutorials/pyspark/pyspark-filtering)

1. Concept Overview The `orderBy(desc("amount"))` transformation sorts the DataFrame in descending order based on the `amount` column, placing the highest numeric value at the very top (first row). The sorted order is `[300, 200, 100]`. Calling `.first()` retrieves the top row, and `["amount"]` accesses the value of the `amount` column in that row, returning `300`.

2. Architecture & Design Principle 1. **Global Sorting vs. Local Sorting**: - `orderBy()` (or `sort()`) performs a **global sort** across all partitions. - Spark samples the data using a `RangePartitioner` so that partition 0 receives the highest range of values, partition 1 receives the next range, and so forth. - Within each partition, rows are sorted in descending order. 2. **Action Retrieval**: `.first()` fetches the first row of partition 0 to the driver node without collecting the rest of the dataset.

3. Step-by-Step Code / Logic Execution 1. Input values: `[100, 200, 300]`. 2. `orderBy(desc("amount"))` sorts the rows descending: - Row 1: 300 - Row 2: 200 - Row 3: 100 3. `.first()` returns `Row(amount=300)`. 4. `["amount"]` extracts the value: `300`.

4. Why the Correct Answer Is Optimal Option B is correct because: - Descending order places the maximum value (300) in the first position. - Accessing the first record's `amount` attribute returns 300.

5. Why Alternative Options Fail - Option A (100): Incorrect; 100 would be returned if sorted in ascending order (`asc("amount")`). - Option C (200): Incorrect; 200 is the median/second value. - Option D (NULL): Incorrect; no null values exist in the input dataset.

6. Edge Cases & Production Pitfalls - **Null Placement in Descending Sorts**: In Spark, `desc("amount")` defaults to `NULLS FIRST`. If the dataset contains `NULL`, the first row would be `NULL` unless explicitly specified as `col("amount").desc_nulls_last()`. - **Performance Cost of Global Sorts**: Sorting a 1-billion-row DataFrame globally is extremely expensive. If your only goal is finding the maximum value, avoid `orderBy().first()` and use `df.agg(max("amount")).first()[0]`, which calculates the maximum in a single linear pass with no global shuffle!

7. Performance & Optimization Considerations - Prefer `df.agg(max("amount"))` over `df.orderBy(desc("amount")).first()` for finding maximums; `agg(max())` requires only partition-level scalar reductions rather than full data shuffling.

8. Reliability & Operational Best Practices - Always specify null ordering explicitly (`.desc_nulls_last()`) when sorting columns that may contain null values to ensure predictable results.

9. Real-World Architectural Scenario An auction platform determines the winning bid for completed items. Querying `bids_df.filter(col("item_id") == 42).orderBy(col("bid_amount").desc()).first()` identifies the highest bidder and winning purchase amount.

10. Official Documentation & Learning Resources - Sorting Operations: [PySpark Sorting Sampling & Limit Guide](/tutorials/pyspark/sorting-sampling-limit-pyspark)

1. Concept Overview This query performs a two-step aggregation. First, `df.groupBy("customer_id").count()` groups the DataFrame by `customer_id` and computes the number of records per customer. In the input dataset, there are two distinct customers: 'C1' (with 2 transactions) and 'C2' (with 1 transaction). This produces a grouped intermediate DataFrame with 2 rows. Calling the terminal action `.count()` on that grouped DataFrame counts the number of grouped rows, returning `2`.

2. Architecture & Design Principle 1. **GroupBy Transformation vs. DataFrame Action**: - The first `.count()` is a **transformation method on a GroupedData object**, producing a new DataFrame: `RelationalGroupedDataset.count()`. - The second `.count()` is an **action on the resulting DataFrame**, triggering physical execution and returning an integer scalar representing the total row count. 2. **Intermediate Schema**: - Intermediate DataFrame:

text
     +-----------+-----+
     |customer_id|count|
     +-----------+-----+
     |         C1|    2|
     |         C2|    1|
     +-----------+-----+

• Row count of this intermediate table = 2.

3. Step-by-Step Code / Logic Execution 1. Original input records: - C1, 100 - C1, 200 - C2, 300 2. `df.groupBy("customer_id").count()` aggregates by `customer_id`: - Customer C1 has 2 records. - Customer C2 has 1 record. - Resulting DataFrame has 2 rows. 3. Terminal `.count()` counts the rows in the resulting DataFrame: - Output = `2`.

4. Why the Correct Answer Is Optimal Option A is correct because: - The dataset contains exactly 2 distinct customer IDs (C1 and C2). - The intermediate grouped DataFrame contains exactly 2 rows, which `.count()` tallies as 2.

5. Why Alternative Options Fail - Option B (3): Incorrect; 3 is the original row count before grouping. - Option C (4): Incorrect; no data multiplication occurred. - Option D (1): Incorrect; there are two distinct customer groups, not one.

6. Edge Cases & Production Pitfalls - **Method Overloading Confusion**: Developers often confuse the two `count()` methods: - `df.groupBy("col").count()`: Returns a DataFrame. - `df.count()`: Returns an integer. - **Null Grouping Keys**: In PySpark, `groupBy()` groups `NULL` keys into their own group. If a row had `customer_id = NULL`, a third row with `customer_id = NULL` would appear in the grouped result.

7. Performance & Optimization Considerations - Grouping requires a shuffle across the cluster. To find the number of distinct customers without computing group counts, use `df.select("customer_id").distinct().count()`.

8. Reliability & Operational Best Practices - When grouping by keys, monitor Spark UI shuffle read/write sizes to detect data skew if a single customer possesses millions of transactions.

9. Real-World Architectural Scenario A financial compliance service verifies the number of active distinct corporate entities that conducted business during a monthly reporting cycle. Calling `df.groupBy("entity_id").count().count()` provides the audited total number of unique reporting entities.

10. Official Documentation & Learning Resources - GroupBy and Aggregations: [PySpark GroupBy & Aggregate Operations Guide](/tutorials/pyspark/pyspark-aggregation)