Q: Discuss execution flow in a distributed environment: How does Spark handle fault tolerance and parallel execution? What role does the DAG (Directed Acyclic Graph) play in optimizing query execution?

Parallel execution flow: 1. Driver submits an Application → SparkContext negotiates resources with the cluster manager (YARN/Kubernetes) 2. Executors are launched on worker nodes 3. Each Action triggers a Job submission 4. Spark divides the job into Stages (split at wide transformation boundaries) 5. Each stage's tasks run in parallel — one task per partition, distributed across available executor cores 6. Stages run sequentially when one depends on another's shuffle output Fault tolerance: Spark achieves fault tolerance through RDD lineage — the DAG records every transformation applied to the data If an executor or partition fails, Spark recomputes only the lost partitions by replaying the lineage from the last checkpoint or source No full job restart required — lost partitions are recomputed on healthy nodes persist() / checkpoint() materialize intermediate results, shortening recomputation chains for iterative workloads Role of the DAG: The DAG is the logical execution plan — a directed graph where nodes are RDD/DataFrame transformations and edges are dependencies The DAG Scheduler converts the DAG into physical stages by identifying shuffle boundaries Before execution, the Catalyst optimizer traverses the DAG to: Push filters down (predicate pushdown) Eliminate redundant projections (column pruning) Reorder joins based on table statistics Fuse adjacent narrow transformations into single pipeline stages This optimization pass happens transparently, making hand written Spark SQL nearly as efficient as a tuned execution plan

Q: What is the purpose of default_args, and why do we set catchup=False?

default args in Airflow is a dictionary of default parameters applied to all tasks in a DAG, reducing boilerplate when defining individual operators. These values are inherited by each operator in the DAG unless overridden at the task level. This is DRY (Don't Repeat Yourself) configuration. catchup=False — why it's important: When a DAG has a start date in the past and catchup=True (the default), Airflow will backfill — create and run all missed DAG runs from start date to the current date. Example : If you deploy a daily DAG with start date='2024 01 01' today (April 2024) with catchup=True , Airflow immediately queues ~90 historical DAG runs. Why catchup=False is typically preferred: Prevents an accidental flood of backfill runs that can overwhelm your pipeline and downstream systems Avoids re processing historical data unintentionally Most production pipelines are designed to process current data, not historical backfills If backfilling is needed, it should be done explicitly via airflow dags backfill

Question 1

As a Data Engineer, analyze the Spark transformations and actions used in the following PySpark workflow and explain their impact on performance and execution.

Accepted Answer

When analyzing a PySpark workflow, you must distinguish transformations (lazy, build the DAG) from actions (trigger execution). Transformations (lazy — no execution until an action is called): df.read , filter , select , withColumn , groupBy , join , repartition , cache Each transformation adds a node to the Directed Acyclic Graph (DAG) Spark batches and optimizes the whole DAG via the Catalyst optimizer before executing Actions (trigger execution): .show() , .count() , .collect() , .write , .toPandas() Calling an action submits a Job to the Spark scheduler Performance impact analysis: filter is cheap — single partition operation, no network I/O join and groupBy each trigger a shuffle (data redistributed across network) Two wide transformations = two shuffle stages = primary performance bottleneck Optimizations: broadcast small DataFrames, enable AQE, filter before joining

Question 2

Identify transformations and actions in the given PySpark code.

Accepted Answer

Quick reference — Transformations vs Actions: Transformations (return a new DataFrame/RDD, lazy): Operation Type Notes filter() / where() Narrow No shuffle select() / withColumn() Narrow Column operations map() / flatMap() Narrow RDD operations union() Narrow No shuffle join() Wide Shuffles both sides groupBy() / groupByKey() Wide Shuffle distinct() Wide Shuffle repartition() Wide Full shuffle coalesce() Narrow Merge, no shuffle orderBy() / sort() Wide Global sort Actions (trigger execution, return results): Operation Notes .show(n) Print n rows .count() Return row count .collect() Return all rows to driver .take(n) Return first n rows .first() Return first row .write.parquet(...) Write to storage .toPandas() Convert to Pandas DataFrame .foreach() Apply function to each row Key rule : If it returns a DataFrame → transformation. If it returns a value or writes output → action.

Question 3

What are narrow and wide transformations?

Accepted Answer

Narrow transformations : Each output partition depends on data from exactly one input partition. No data movement between executors — all processing is local. Examples: filter , map , flatMap , select , withColumn , union , coalesce No shuffle → no stage boundary → fast, no network I/O Multiple consecutive narrow transformations are pipelined in a single stage Wide transformations : Each output partition depends on data from multiple input partitions. Requires a shuffle — data is redistributed across the cluster. Examples: groupBy , join , distinct , repartition , orderBy , reduceByKey Each wide transformation creates a stage boundary in the DAG Shuffle involves: serialize → write to disk → transfer over network → deserialize Why it matters : Wide transformations are the primary source of Spark performance issues. Minimizing them (via broadcast joins, pre partitioning, early filtering) is the most impactful optimization strategy.

Question 4

Why is filtering (df.filter(col('salary') > 10000)) considered a narrow transformation while join is a wide transformation?

Accepted Answer

filter is narrow because evaluating salary 10000 for a row requires only the data within that single row, which lives in a single partition. No information from any other partition is needed. The operation is perfectly local to each executor — each executor filters its own partitions independently with zero network communication. join is wide because to correctly match rows from the left DataFrame with rows from the right DataFrame on a key, rows with the same key must be co located on the same executor . Since related rows can start in different partitions on different executors, Spark must shuffle both DataFrames — redistributing all rows by their join key — so matching rows end up together. Summary : The distinguishing factor is data locality . Narrow = each partition is self contained for the operation. Wide = the operation requires data from multiple partitions to be combined, necessitating a shuffle across the network.

Question 5

How does Spark execute transformations lazily? When does actual computation happen? Why does Spark not execute transformations immediately but wait until an action (like .show()) is called?

Accepted Answer

Lazy evaluation means Spark records what transformations to perform (building a DAG of operations) but does not execute any computation until an action is called. How it works: When computation actually happens: Computation triggers when an action is called: .show() , .count() , .collect() , .write. (...) , .toPandas() . Why lazy evaluation is beneficial: 1. Whole plan optimization : By seeing the complete chain of transformations before executing, the Catalyst optimizer can reorder, merge, or eliminate operations. For example, it pushes filter before a join even if you wrote the join first in code — scanning less data 2. Avoid unnecessary computation : If you build a transformation chain but never call an action on some branches (e.g., in a conditional code path), no work is done for that branch 3. Pipeline optimization : Consecutive narrow transformations are merged into a single stage, processed in one pass over the data rather than materializing intermediate results 4. Stage planning : Spark can plan all shuffle boundaries upfront, minimizing data movement Lazy evaluation is what allows Spark to be dramatically more efficient than executing each operation eagerly.

Question 6

Explain the impact of the join operation: How does the shuffle stage impact performance? How can you optimize join performance in Spark (e.g., using broadcast joins)?

Accepted Answer

Shuffle stage impact on performance: A shuffle involves three expensive phases: 1. Map phase : Each executor writes its output partitioned by join key to local disk (shuffle write) 2. Network transfer : Shuffle files are pulled by destination executors over the network (the most expensive part) 3. Reduce phase : Receiving executors read and sort/hash the data to perform the join (shuffle read) Costs include: disk I/O, network bandwidth, serialization/deserialization, and memory pressure (shuffle spills to disk when memory is insufficient). A poorly managed shuffle can cause OOM errors, straggler tasks (data skew), and overall job slowness. Join optimization strategies: 1. Broadcast Hash Join (most impactful): If one table fits in executor memory (< spark.sql.autoBroadcastJoinThreshold , default 10MB), broadcast it to every executor. The join becomes local — zero network shuffle. 2. Partition co location : Pre partition both tables by the join key and write to storage. On re read, co partitioned data can be joined without shuffling. 3. Bucketing : Use bucketBy to store tables pre sorted and bucketed on join keys — Spark can skip the shuffle for bucket compatible joins. 4. Filter before join : Reduce the size of both inputs as much as possible before the join. 5. Enable AQE : spark.sql.adaptive.enabled=true — AQE can automatically switch to a broadcast join at runtime if it detects a table became small after filtering.

Question 7

Discuss execution flow in a distributed environment: How does Spark handle fault tolerance and parallel execution? What role does the DAG (Directed Acyclic Graph) play in optimizing query execution?

Accepted Answer

Parallel execution flow: 1. Driver submits an Application → SparkContext negotiates resources with the cluster manager (YARN/Kubernetes) 2. Executors are launched on worker nodes 3. Each Action triggers a Job submission 4. Spark divides the job into Stages (split at wide transformation boundaries) 5. Each stage's tasks run in parallel — one task per partition, distributed across available executor cores 6. Stages run sequentially when one depends on another's shuffle output Fault tolerance: Spark achieves fault tolerance through RDD lineage — the DAG records every transformation applied to the data If an executor or partition fails, Spark recomputes only the lost partitions by replaying the lineage from the last checkpoint or source No full job restart required — lost partitions are recomputed on healthy nodes persist() / checkpoint() materialize intermediate results, shortening recomputation chains for iterative workloads Role of the DAG: The DAG is the logical execution plan — a directed graph where nodes are RDD/DataFrame transformations and edges are dependencies The DAG Scheduler converts the DAG into physical stages by identifying shuffle boundaries Before execution, the Catalyst optimizer traverses the DAG to: Push filters down (predicate pushdown) Eliminate redundant projections (column pruning) Reorder joins based on table statistics Fuse adjacent narrow transformations into single pipeline stages This optimization pass happens transparently, making hand written Spark SQL nearly as efficient as a tuned execution plan

Question 8

What is the purpose of default_args, and why do we set catchup=False?

Accepted Answer

default args in Airflow is a dictionary of default parameters applied to all tasks in a DAG, reducing boilerplate when defining individual operators. These values are inherited by each operator in the DAG unless overridden at the task level. This is DRY (Don't Repeat Yourself) configuration. catchup=False — why it's important: When a DAG has a start date in the past and catchup=True (the default), Airflow will backfill — create and run all missed DAG runs from start date to the current date. Example : If you deploy a daily DAG with start date='2024 01 01' today (April 2024) with catchup=True , Airflow immediately queues ~90 historical DAG runs. Why catchup=False is typically preferred: Prevents an accidental flood of backfill runs that can overwhelm your pipeline and downstream systems Avoids re processing historical data unintentionally Most production pipelines are designed to process current data, not historical backfills If backfilling is needed, it should be done explicitly via airflow dags backfill

Synechron Interview Questions