Q: Write a Python function that validates incoming JSON data against a schema and logs invalid records to a separate file.

Use the jsonschema library for validation and plain file writes for invalid records: jsonschema.validate() raises ValidationError on the first violation. Use Draft7Validator if you need all errors per record. In production pipelines (Airflow / Spark), the invalid records file should land in GCS/S3 with a timestamped key for later auditing.

Q: Given a list of transaction dictionaries, write a Python script to calculate total spend per user and output as a JSON.

defaultdict(float) avoids a KeyError check on every iteration. Always round() monetary values to 2 decimal places in production to avoid floating point drift. For large datasets, prefer Pandas ( groupby('user id')['amount'].sum() ) or PySpark for distributed aggregation.

Q: How does schedule_interval='@daily' affect DAG execution?

@daily is an Airflow preset equivalent to the cron expression 0 0 — it schedules the DAG to run once per day at midnight UTC . Important Airflow scheduling nuances: Execution date vs run date : The DAG run with execution date = 2024 01 01 actually runs on 2024 01 02 00:00 UTC . Airflow uses the end of the interval as the trigger. Catchup : If catchup=True (default), Airflow backfills all missed runs from start date to now. Set catchup=False for new DAGs that don't need historical backfill. Data interval : In Airflow 2.x, data interval start and data interval end replace execution date for clarity in templated fields. Timezone : Unless a timezone aware start date is provided via pendulum , all times are treated as UTC. Best practice: always set start date to a fixed past date and catchup=False unless historical backfill is explicitly required.

Q: What is the difference between PythonOperator and BashOperator?

Both are Airflow task operators but differ in execution context: Feature PythonOperator BashOperator Executes A Python callable A bash shell command or script Context Runs in the worker's Python environment Spawns a subprocess XCom Native via return or ti.xcom push Must parse stdout or use xcom push flag Dependencies Any installed Python library Any CLI tool or shell script Error handling Python exceptions captured cleanly Non zero exit codes raise AirflowException When to use each: PythonOperator — for data transformations, API calls, or business logic in Python. BashOperator — for running shell scripts, CLI tools ( dbt run , gsutil , spark submit ), or legacy bash workflows. Note : In Airflow 2.x, prefer the @task decorator (TaskFlow API) over PythonOperator for cleaner, more testable DAG code.

Q: How does Airflow schedule and execute tasks?

Airflow uses a scheduler loop that continuously parses DAG files, determines which DAG runs are due, and submits tasks to an executor: 1. Scheduler reads all DAG files in AIRFLOW HOME/dags/ at a configurable interval. 2. It evaluates which DAG runs are due based on schedule interval and creates DagRun records in the metadata database. 3. For each DagRun , it evaluates task dependencies (the DAG graph) and marks tasks whose upstream dependencies are satisfied as scheduled . 4. Scheduled tasks are submitted to the Executor (LocalExecutor, CeleryExecutor, KubernetesExecutor). 5. The executor hands tasks to Workers , which import the DAG file, instantiate the operator, and call execute(context) . 6. Task state (running, success, failed, retrying) is updated in the metadata DB (PostgreSQL or MySQL). Key components: Scheduler , Webserver , Executor , Workers , Metadata DB . The webserver is stateless and only reads from the metadata DB to render the UI.

Q: The BigQuery query worked fine for months but recently started timing out. How would you troubleshoot it?

A systematic troubleshooting approach: 1. Check query execution plan : Inspect the BigQuery Job information panel to see estimated bytes processed, join order, and stages. 2. Data growth : Use INFORMATION SCHEMA.TABLE STORAGE to verify whether underlying tables have grown significantly in size or row count. 3. Partition/cluster pruning : Confirm that WHERE clause filters actually hit partition columns. A missing WHERE event date BETWEEN ... on a partitioned table causes a full scan. 4. JOIN explosions : Look for fan out joins or accidental cross joins producing far more rows than expected. 5. Slot quota : Check the project's slot utilization in Cloud Monitoring. On on demand pricing, concurrent heavy jobs compete for slots. 6. Recent schema changes : Check if a new REPEATED or STRUCT column was added that inflated row sizes. 7. Caching changes : Verify if a materialized view or query result cache stopped being hit. 8. Remediation : Add partition filters, break the query into steps using temp tables, or upgrade to BigQuery reservations for guaranteed capacity.

Question 1

Write a PySpark code snippet to: Read data from a Parquet file, Filter rows where status = 'FAILED', Group by customer_id and count failures.

Accepted Answer

PySpark handles this with a simple read filter aggregate pipeline: spark.read.parquet() reads Parquet using columnar pushdown, so only the status and customer id columns are scanned. .filter() applies predicate pushdown — Parquet row group statistics allow Spark to skip irrelevant chunks entirely. .groupBy().agg() triggers a wide transformation (shuffle). To reduce shuffle overhead, tune spark.sql.shuffle.partitions and consider bucketing customer id if this aggregation runs repeatedly.

Question 2

How would you optimize a wide transformation like groupBy() or join() on large datasets in PySpark?

Accepted Answer

Wide transformations require shuffling data across partitions, which is the most expensive Spark operation. Key optimizations: Broadcast joins : If one side is small (< ~10 MB default), use broadcast() hint to skip shuffling the large table. Partition tuning : Set spark.sql.shuffle.partitions to ~2–3× the number of cores. The default of 200 causes tiny files on small data or huge partitions on big data. Bucketing : Write tables with bucketBy(n, 'key') so that joins on that key skip the shuffle entirely on future reads. Salting skewed keys : If one key value dominates, add a random salt suffix to split it across multiple reducers. Persist intermediate DFs : If the same DF is used in multiple downstream wide transformations, .cache() it to avoid recomputing upstream. Predicate pushdown and column pruning : Select only needed columns before joins/groupBys to reduce data shuffled over the network.

Question 3

Write a Python function that validates incoming JSON data against a schema and logs invalid records to a separate file.

Accepted Answer

Use the jsonschema library for validation and plain file writes for invalid records: jsonschema.validate() raises ValidationError on the first violation. Use Draft7Validator if you need all errors per record. In production pipelines (Airflow / Spark), the invalid records file should land in GCS/S3 with a timestamped key for later auditing.

Question 4

Given a list of transaction dictionaries, write a Python script to calculate total spend per user and output as a JSON.

Accepted Answer

defaultdict(float) avoids a KeyError check on every iteration. Always round() monetary values to 2 decimal places in production to avoid floating point drift. For large datasets, prefer Pandas ( groupby('user id')['amount'].sum() ) or PySpark for distributed aggregation.

Question 5

How does schedule_interval='@daily' affect DAG execution?

Accepted Answer

@daily is an Airflow preset equivalent to the cron expression 0 0 — it schedules the DAG to run once per day at midnight UTC . Important Airflow scheduling nuances: Execution date vs run date : The DAG run with execution date = 2024 01 01 actually runs on 2024 01 02 00:00 UTC . Airflow uses the end of the interval as the trigger. Catchup : If catchup=True (default), Airflow backfills all missed runs from start date to now. Set catchup=False for new DAGs that don't need historical backfill. Data interval : In Airflow 2.x, data interval start and data interval end replace execution date for clarity in templated fields. Timezone : Unless a timezone aware start date is provided via pendulum , all times are treated as UTC. Best practice: always set start date to a fixed past date and catchup=False unless historical backfill is explicitly required.

Question 6

What is the difference between PythonOperator and BashOperator?

Accepted Answer

Both are Airflow task operators but differ in execution context: Feature PythonOperator BashOperator Executes A Python callable A bash shell command or script Context Runs in the worker's Python environment Spawns a subprocess XCom Native via return or ti.xcom push Must parse stdout or use xcom push flag Dependencies Any installed Python library Any CLI tool or shell script Error handling Python exceptions captured cleanly Non zero exit codes raise AirflowException When to use each: PythonOperator — for data transformations, API calls, or business logic in Python. BashOperator — for running shell scripts, CLI tools ( dbt run , gsutil , spark submit ), or legacy bash workflows. Note : In Airflow 2.x, prefer the @task decorator (TaskFlow API) over PythonOperator for cleaner, more testable DAG code.

Question 7

How does Airflow schedule and execute tasks?

Accepted Answer

Airflow uses a scheduler loop that continuously parses DAG files, determines which DAG runs are due, and submits tasks to an executor: 1. Scheduler reads all DAG files in AIRFLOW HOME/dags/ at a configurable interval. 2. It evaluates which DAG runs are due based on schedule interval and creates DagRun records in the metadata database. 3. For each DagRun , it evaluates task dependencies (the DAG graph) and marks tasks whose upstream dependencies are satisfied as scheduled . 4. Scheduled tasks are submitted to the Executor (LocalExecutor, CeleryExecutor, KubernetesExecutor). 5. The executor hands tasks to Workers , which import the DAG file, instantiate the operator, and call execute(context) . 6. Task state (running, success, failed, retrying) is updated in the metadata DB (PostgreSQL or MySQL). Key components: Scheduler , Webserver , Executor , Workers , Metadata DB . The webserver is stateless and only reads from the metadata DB to render the UI.

Question 8

The BigQuery query worked fine for months but recently started timing out. How would you troubleshoot it?

Accepted Answer

A systematic troubleshooting approach: 1. Check query execution plan : Inspect the BigQuery Job information panel to see estimated bytes processed, join order, and stages. 2. Data growth : Use INFORMATION SCHEMA.TABLE STORAGE to verify whether underlying tables have grown significantly in size or row count. 3. Partition/cluster pruning : Confirm that WHERE clause filters actually hit partition columns. A missing WHERE event date BETWEEN ... on a partitioned table causes a full scan. 4. JOIN explosions : Look for fan out joins or accidental cross joins producing far more rows than expected. 5. Slot quota : Check the project's slot utilization in Cloud Monitoring. On on demand pricing, concurrent heavy jobs compete for slots. 6. Recent schema changes : Check if a new REPEATED or STRUCT column was added that inflated row sizes. 7. Caching changes : Verify if a materialized view or query result cache stopped being hit. 8. Remediation : Add partition filters, break the query into steps using temp tables, or upgrade to BigQuery reservations for guaranteed capacity.

Fractal Interview Questions