Data Engineer Interview Questions for Experienced 2025
Advancing to senior data engineering positions requires demonstrating advanced system architecture expertise and scalable infrastructure design beyond foundational skills. Data Engineer Interview Questions for Experienced professionals focus on complex pipeline optimization, distributed systems management, and technical leadership that seasoned engineers encounter.
This interview guide covers Data Engineer Interview Questions for Experienced candidates with multiple years of industry experience, addressing real-time processing frameworks, cloud infrastructure decisions, and cross-team collaboration scenarios. Data Engineer Interview Questions for Experienced professionals will help you showcase your expertise, demonstrate system performance improvements, and prove your readiness for senior data engineering roles in today’s infrastructure-focused landscape.
You can check another Data Engineer Interview guide here: Data Engineer Interview Questions PDF
Table of Contents
Data Engineer Interview Questions for 2 Years Experience
Que. 1 How would you design a data pipeline to handle incremental data loads from a relational database to a data lake, ensuring no data loss and minimal latency?
Answer:
To design an incremental data pipeline, use a combination of change data capture (CDC) and batch processing. Start by identifying a source database (e.g., PostgreSQL) with a timestamp or incremental ID column for tracking changes. Use a tool like Debezium to capture CDC events from the database’s transaction log and stream them to Kafka.
Extract changes by configuring Debezium to publish to a Kafka topic, with each event containing the row’s data, operation (insert/update/delete), and timestamp. In Apache Spark, consume the stream using spark.readStream.format("kafka"), ensuring exactly-once delivery with checkpointing.
Transform the data by applying deduplication logic (e.g., keeping the latest record per primary key) and converting to Parquet for columnar storage. Use mergeInto in Delta Lake to upsert changes into the data lake (e.g., S3), maintaining a history table with is_active flags for updates.
To minimize latency, tune Spark’s trigger interval (e.g., trigger(processingTime="5 seconds")) and ensure sufficient Kafka partitions. Monitor pipeline health with Prometheus metrics and set alerts for lag or errors. This ensures reliable, low-latency incremental loads without data loss.
Que. 2 Write a SQL query to identify duplicate records in a table with columns: user_id, email, signup_date, and return only the latest record per user_id.
Answer:
To identify duplicates and keep the latest record per user_id, use a window function to rank records by signup_date. The query assumes duplicates occur on user_id.
WITH ranked_users AS (
SELECT
user_id,
email,
signup_date,
ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY signup_date DESC) AS rn
FROM users
)
SELECT
user_id,
email,
signup_date
FROM ranked_users
WHERE rn = 1;This assigns rn=1 to the latest record per user_id based on signup_date. The outer query filters for these records, effectively deduplicating while preserving the most recent entry.
Que. 3 How do you handle schema evolution in a data lake when new columns are added to source data, and how do you ensure downstream compatibility?
Answer:
Schema evolution in a data lake requires handling new columns while maintaining compatibility with downstream processes. Use a format like Parquet or Delta Lake, which supports schema-on-read and schema evolution.
When new columns appear, configure ingestion pipelines (e.g., Spark or AWS Glue) to merge schemas dynamically using mergeSchema=true. For Delta Lake, enable schema.autoMerge=true to automatically incorporate new columns during writes.
For downstream compatibility, maintain a schema registry (e.g., Confluent Schema Registry for Kafka or AWS Glue Data Catalog) to track versions. Enforce forward-compatible changes by adding nullable columns and avoiding deletions. Transform data in ETL jobs to align with expected schemas, using default values for missing columns (e.g., coalesce(col("new_col"), lit(null))).
Test compatibility by running validation jobs against a staging environment before production. Log schema changes to monitor drift, ensuring analytics tools like Redshift or BigQuery handle new columns gracefully.
Que. 4 Write a PySpark script to calculate the moving average of sales over a 7-day window for each product in a dataset with columns: product_id, sale_date, sales_amount.
Answer:
To compute a 7-day moving average, use Spark’s window functions to define a sliding window over sale_date.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, window
from pyspark.sql.window import Window
spark = SparkSession.builder.appName("MovingAverage").getOrCreate()
df = spark.read.parquet("sales_data")
window_spec = Window.partitionBy("product_id").orderBy(col("sale_date").cast("timestamp")).rangeBetween(-6*24*60*60, 0)
df_with_ma = df.withColumn("moving_avg", avg("sales_amount").over(window_spec))
df_with_ma.write.format("parquet").save("output_path")
spark.stop()The rangeBetween defines a 7-day window (6 days prior to current day, in seconds). The avg function computes the moving average per product_id, ordered by sale_date. Results are saved in Parquet for downstream use.
Que. 5 What strategies would you use to optimize storage costs in a cloud-based data lake like AWS S3 for large-scale datasets?
Answer:
Optimizing storage costs in a data lake involves efficient formats, partitioning, and lifecycle policies. Use columnar formats like Parquet or ORC to reduce size through compression and columnar pruning. Enable snappy compression (spark.sql.parquet.compression.codec=snappy) for a balance of speed and size.
Partition data by high-cardinality columns like date or region to enable partition pruning, reducing scan costs (e.g., year=2025/month=08). Avoid over-partitioning to prevent small file issues; consolidate with coalesce or compaction jobs.
Implement S3 lifecycle policies to transition older data to cheaper tiers like S3 Glacier (e.g., after 90 days) or delete unused data. Use AWS S3 Storage Lens to monitor usage and identify wasteful patterns.
Run periodic compaction jobs with Delta Lake to merge small files and vacuum old versions (delta.vacuum(168) for 7-day retention). These steps minimize storage costs while maintaining query performance.
Que. 6 How would you implement a retry mechanism in an Airflow DAG for a task that fetches data from an unreliable API?
Answer:
To implement retries in an Airflow DAG, configure the task’s retry parameters and handle transient failures gracefully. Define the DAG with a PythonOperator for the API task.
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import requests
def fetch_api_data():
try:
response = requests.get("https://api.example.com/data", timeout=10)
response.raise_for_status()
# Process data
except requests.exceptions.RequestException as e:
raise AirflowFailException(f"API call failed: {e}")
with DAG(
dag_id="api_fetch",
start_date=datetime(2025, 1, 1),
schedule_interval="@daily",
default_args={
"retries": 3,
"retry_delay": timedelta(minutes=5),
},
) as dag:
fetch_task = PythonOperator(
task_id="fetch_api_data",
python_callable=fetch_api_data,
)Set retries=3 and retry_delay=timedelta(minutes=5) in default_args. The fetch_api_data function uses requests with a timeout and raises AirflowFailException for retryable failures. Monitor retries via Airflow logs and set alerts for persistent failures.
Que. 7 Explain the difference between Apache Kafka and Apache Pulsar, and when would you choose one over the other for a streaming pipeline?
Answer:
Kafka is a distributed streaming platform with a log-based architecture, excelling in high-throughput, durable message storage. Pulsar, also distributed, uses a tiered architecture separating compute (brokers) from storage (BookKeeper), offering multi-tenancy and flexible scaling.
Kafka is simpler to deploy and widely adopted, ideal for linear pipelines with large-scale event streaming (e.g., log aggregation). Its partitioning model suits high-throughput use cases but lacks native multi-tenancy.
Pulsar supports multi-tenancy and dynamic topic partitioning, making it better for complex, multi-team environments or when needing geo-replication. Its tiered storage allows cost-efficient retention of historical data.
Choose Kafka for simpler, high-throughput pipelines with mature ecosystem integration (e.g., Spark, Flink). Choose Pulsar for multi-tenant setups or when needing advanced features like function-based processing or tiered storage.
Que. 8 Write a SQL query to calculate the churn rate of customers per month from a subscriptions table with columns: user_id, subscription_start, subscription_end.
Answer:
Churn rate is the percentage of customers who end their subscription in a month. Assume subscription_end is null for active subscriptions.
WITH monthly_data AS (
SELECT
DATE_TRUNC('month', subscription_start) AS month,
COUNT(DISTINCT user_id) AS total_users,
COUNT(DISTINCT CASE WHEN DATE_TRUNC('month', subscription_end) = DATE_TRUNC('month', subscription_start) THEN user_id END) AS churned_users
FROM subscriptions
GROUP BY DATE_TRUNC('month', subscription_start)
)
SELECT
month,
total_users,
churned_users,
(churned_users::FLOAT / total_users) * 100 AS churn_rate
FROM monthly_data
ORDER BY month;This groups by month, counts total and churned users (where subscription_end matches the month), and computes the rate. Adjust DATE_TRUNC for your database’s syntax (e.g., MONTH(subscription_start) for SQL Server).
Que. 9 How do you ensure data consistency in a distributed data pipeline that processes data across multiple systems like Kafka, Spark, and a data warehouse?
Answer:
Ensuring data consistency requires idempotency, exactly-once semantics, and monitoring. Use Kafka’s idempotent producer (enable.idempotence=true) to avoid duplicate messages during retries. For Spark, enable checkpointing in Structured Streaming (writeStream.checkpointLocation) to guarantee exactly-once delivery to sinks like Delta Lake.
Implement transactional writes in the data warehouse (e.g., Snowflake’s MERGE or BigQuery’s DML) to prevent partial updates. Use a unique transaction ID or primary key to deduplicate records during ingestion.
Track data lineage with tools like Apache Atlas or OpenLineage to verify source-to-target consistency. Schedule reconciliation jobs to compare record counts and checksums across systems. For example, compute a hash of key columns in Spark and match against the warehouse.
Monitor with alerts for discrepancies and log errors to ELK. This ensures no data loss or duplication across the pipeline.
Que. 10 Write a Python script to validate a CSV file’s schema against an expected structure before processing in a pipeline.
Answer:
Validate the CSV’s columns and data types against a predefined schema using Pandas.
import pandas as pd
from pandas.api.types import is_integer_dtype, is_float_dtype, is_string_dtype
expected_schema = {
"user_id": is_integer_dtype,
"name": is_string_dtype,
"sales": is_float_dtype,
"date": is_string_dtype # Assuming date is string, later convertible
}
def validate_csv(file_path):
df = pd.read_csv(file_path)
actual_columns = set(df.columns)
expected_columns = set(expected_schema.keys())
if actual_columns != expected_columns:
raise ValueError(f"Column mismatch: Expected {expected_columns}, got {actual_columns}")
for col, dtype_check in expected_schema.items():
if not dtype_check(df[col]):
raise ValueError(f"Invalid type for {col}: Expected {dtype_check.__name__}, got {df[col].dtype}")
print("Schema validation passed")
return df
df = validate_csv("input.csv")This checks for missing/extra columns and verifies data types. Extend with additional checks like value ranges or regex for dates if needed.
Guide for Freshers: Data Engineer Interview Questions for Freshers
Data Engineer Interview Questions for 3 Years Experience
Que. 11 How would you troubleshoot a Spark job that fails due to out-of-memory errors on executors?
Answer:
First, check Spark UI for executor metrics, identifying high memory usage or garbage collection times. Common causes include large shuffles, skewed data, or unpersisted DataFrames.
Increase executor memory (spark.executor.memory, e.g., 8g) and cores (spark.executor.cores, e.g., 4) if resources allow. Enable spark.memory.fraction (default 0.6) to allocate more memory for execution over caching.
Address data skew by salting keys or repartitioning (df.repartition("key")). For shuffles, reduce data movement with broadcast joins for small tables or increase spark.sql.shuffle.partitions (e.g., 200) to spread load.
Persist DataFrames with MEMORY_AND_DISK to spill to disk, avoiding recomputation. Use EXPLAIN to inspect the query plan for inefficient operations like wide transformations. If spills persist, switch to a more efficient file format like Parquet to reduce I/O.
Log memory metrics to Prometheus and re-run with adjusted configs.
Que. 12 What is the purpose of a data catalog in a data engineering ecosystem, and how would you implement one using AWS Glue?
Answer:
A data catalog organizes metadata about datasets (schema, location, format) to enable discovery, governance, and lineage tracking. It centralizes metadata for data lakes, warehouses, and pipelines, aiding compliance and collaboration.
In AWS Glue, implement a catalog by crawling S3 data lakes or databases. Configure a Glue Crawler with an IAM role, specifying data sources (e.g., S3 paths like s3://bucket/prefix/). Set the crawler to detect schemas for formats like Parquet or CSV and update the Glue Data Catalog.
Schedule crawlers to handle schema changes and use classifiers for custom formats. Integrate with Lake Formation for access control, tagging sensitive data (e.g., PII). Query the catalog via Athena or integrate with Redshift Spectrum for analytics.
Monitor crawler runs via CloudWatch and ensure naming conventions for tables to simplify discovery. This creates a scalable, governed catalog for the ecosystem.
Que. 13 Write a SQL query to find the top 5 customers by total order value who placed orders in the last 30 days, using tables: orders (order_id, customer_id, order_date, amount) and customers (customer_id, customer_name).
Answer:
Join the tables, filter for recent orders, and aggregate by customer.
SELECT
c.customer_id,
c.customer_name,
SUM(o.amount) AS total_order_value
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
WHERE o.order_date >= CURRENT_DATE - INTERVAL '30 days'
GROUP BY c.customer_id, c.customer_name
ORDER BY total_order_value DESC
LIMIT 5;This sums amount per customer for orders within 30 days, joins with customers for names, and limits to the top 5 by value. Use DATEADD or equivalent for SQL Server.
Que. 14 How do you handle late-arriving data in a streaming pipeline built with Apache Flink or Spark Streaming?
Answer:
Late-arriving data in streaming pipelines requires event-time processing and watermarking. In Spark Structured Streaming, define a watermark on the event-time column, e.g., df.withWatermark("event_time", "10 minutes"), allowing late data within 10 minutes to be processed.
Apply windowed aggregations like groupBy(window("event_time", "1 hour")) to bucket data. Spark updates results as late data arrives until the watermark expires. Write to a sink like Delta Lake with update output mode to reflect changes.
In Flink, use WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMinutes(10)) to handle late events. Assign timestamps with assignTimestampsAndWatermarks and process in event-time windows. Flink’s state backend (e.g., RocksDB) manages late updates.
Monitor watermark progress via metrics and adjust thresholds based on data latency patterns to balance completeness and performance.
Que. 15 Write a Python script using boto3 to upload a local file to an S3 bucket and set a lifecycle policy to move it to Glacier after 90 days.
Answer:
Use boto3 to upload and configure lifecycle rules.
import boto3
from botocore.exceptions import ClientError
s3_client = boto3.client("s3")
bucket_name = "my-data-lake"
file_path = "data.csv"
s3_key = "uploads/data.csv"
# Upload file
try:
s3_client.upload_file(file_path, bucket_name, s3_key)
print(f"Uploaded {file_path} to s3://{bucket_name}/{s3_key}")
except ClientError as e:
print(f"Error uploading: {e}")
# Set lifecycle policy
lifecycle_policy = {
"Rules": [
{
"ID": "MoveToGlacier",
"Status": "Enabled",
"Filter": {"Prefix": "uploads/"},
"Transitions": [
{"Days": 90, "StorageClass": "GLACIER"}
]
}
]
}
try:
s3_client.put_bucket_lifecycle_configuration(
Bucket=bucket_name,
LifecycleConfiguration=lifecycle_policy
)
print("Lifecycle policy applied")
except ClientError as e:
print(f"Error setting lifecycle: {e}")This uploads data.csv to S3 and sets a rule to transition files in the uploads/ prefix to Glacier after 90 days. Ensure IAM permissions allow s3:PutObject and s3:PutLifecycleConfiguration.
Que. 16 How would you secure sensitive data (e.g., PII) in a data pipeline processing customer data from Kafka to a data warehouse?
Answer:
Securing PII involves encryption, access control, and auditing. In Kafka, enable SSL for data in transit (security.protocol=SSL) and use SASL for authentication. Encrypt PII fields at the source using libraries like pyspark.sql.functions.encrypt or AWS KMS before publishing to Kafka.
In the pipeline (e.g., Spark), mask sensitive fields (e.g., replace SSN with substring(col("ssn"), 0, 4) || "XXXX"). Store data in encrypted storage like S3 with server-side encryption (SSE-KMS). Use Delta Lake’s row-level security to restrict access.
In the warehouse (e.g., Snowflake), apply dynamic data masking and role-based access control (RBAC). Encrypt columns with ENCRYPT functions and grant access only to authorized roles.
Audit access with tools like AWS CloudTrail or Snowflake’s query history. Tag PII data in a catalog (e.g., AWS Glue) and monitor for compliance using AWS Macie. This ensures end-to-end security.
Que. 17 Write a SQL query to find the longest streak of consecutive days with orders for each customer, using an orders table (order_id, customer_id, order_date).
Answer:
Use a self-join or window functions to identify consecutive days.
WITH date_diffs AS (
SELECT
customer_id,
order_date,
LAG(order_date) OVER (PARTITION BY customer_id ORDER BY order_date) AS prev_date,
CASE
WHEN order_date = LAG(order_date) OVER (PARTITION BY customer_id ORDER BY order_date) + INTERVAL '1 day'
THEN 0 ELSE 1
END AS streak_reset
FROM orders
),
streak_groups AS (
SELECT
customer_id,
order_date,
SUM(streak_reset) OVER (PARTITION BY customer_id ORDER BY order_date) AS streak_id
FROM date_diffs
),
streak_lengths AS (
SELECT
customer_id,
streak_id,
COUNT(*) AS streak_length
FROM streak_groups
GROUP BY customer_id, streak_id
)
SELECT
customer_id,
MAX(streak_length) AS longest_streak
FROM streak_lengths
GROUP BY customer_id
ORDER BY customer_id;This marks streak resets when dates aren’t consecutive, groups by streak, and finds the longest streak per customer.
Que. 18 How do you optimize query performance in a data warehouse like Snowflake or BigQuery for large-scale analytical queries?
Answer:
Optimize by leveraging the warehouse’s architecture and tuning queries. In Snowflake, use clustering keys on frequently filtered columns (e.g., ALTER TABLE sales CLUSTER BY (order_date)) to reduce scan volume. In BigQuery, partition tables by date or range (e.g., PARTITION BY DATE(order_date)) and cluster on join keys.
Write efficient SQL: avoid SELECT *, use aggregations early, and filter with WHERE before joins. Use materialized views for repetitive queries. In Snowflake, enable auto-suspend to save costs and scale compute with warehouse sizing (e.g., ALTER WAREHOUSE SET WAREHOUSE_SIZE=MEDIUM).
Monitor query performance via Snowflake’s Query Profile or BigQuery’s execution details to identify spills or skewed joins. Cache results for repeated queries and use BI tools like Looker to leverage cached data. This reduces latency and compute costs.
Que. 19 Write a PySpark script to join two large datasets and handle missing values before writing to a Delta Lake table.
Answer:
Join datasets and impute missing values using Spark’s DataFrame API.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, coalesce, lit
spark = SparkSession.builder.appName("JoinAndImpute").getOrCreate()
df1 = spark.read.parquet("s3://bucket/customers")
df2 = spark.read.parquet("s3://bucket/orders")
joined_df = df1.join(df2, "customer_id", "left")
imputed_df = joined_df.withColumn("order_amount", coalesce(col("order_amount"), lit(0.0))) \
.withColumn("customer_name", coalesce(col("customer_name"), lit("Unknown")))
imputed_df.write.format("delta").mode("overwrite").save("s3://bucket/delta/joined_table")
spark.stop()This performs a left join, imputes order_amount with 0.0 and customer_name with “Unknown,” and writes to a Delta Lake table with overwrite mode for idempotency.
Que. 20 How would you implement data versioning in a data lake, and what tools or formats would you use to manage historical data?
Answer:
Data versioning tracks changes to datasets, enabling point-in-time queries. Use Delta Lake for versioning due to its time travel and ACID support. Store data in Parquet on S3, with Delta Lake’s transaction log capturing inserts, updates, and deletes.
Enable versioning by writing to a Delta table (df.write.format("delta").save("s3://bucket/table")). Query historical versions using AS OF syntax (e.g., SELECT * FROM delta.s3://bucket/tableVERSION AS OF 5) or timestamp-based queries.
For management, use delta.vacuum() to remove old versions after a retention period (e.g., 7 days). Integrate with a catalog like AWS Glue to track table versions. Use Spark or Databricks jobs to compact small files and optimize performance (OPTIMIZE table_name).
This approach ensures efficient versioning with minimal overhead, supporting audit and rollback needs.
Data Engineer Interview Questions for 5 Years Experience
Que 21. How would you design a scalable ETL pipeline to ingest real-time streaming data from Kafka into a data lake using Apache Spark, and then transform it for querying in a data warehouse?
Answer:
A scalable ETL pipeline for real-time streaming data involves several key components to handle high throughput, fault tolerance, and efficient processing. Start by using Kafka as the message broker for ingesting data from sources like IoT devices or application logs. Apache Spark Structured Streaming can consume data from Kafka topics, providing exactly-once semantics and integration with Spark’s processing engine.
For the extraction phase, configure Spark to read from Kafka using the spark.readStream.format("kafka") method, specifying bootstrap servers, topics, and starting offsets. To ensure scalability, partition the Kafka topics appropriately based on data volume and use Spark’s dynamic allocation for executors.
In the transformation phase, apply operations like filtering invalid records, aggregating metrics (e.g., using groupBy and agg), or joining with static datasets. Use watermarking to handle late-arriving data, for example, withWatermark("timestamp", "10 minutes") to bound state and manage event-time processing. Persist intermediate results if needed using cache() for repeated computations.
For loading, write the processed stream to a data lake like S3 in Parquet format via writeStream.format("parquet") with checkpointing enabled for fault recovery. To make it queryable in a warehouse like Snowflake or BigQuery, use a batch job or Spark’s foreachBatch to upsert into the warehouse tables, leveraging Delta Lake for ACID transactions if using Databricks.
To monitor and scale, integrate with tools like Prometheus for metrics and set auto-scaling policies based on CPU utilization. This design handles increasing data volumes by distributing load across Spark clusters and Kafka partitions.
Que 22. Write a SQL query to find the top 3 products by revenue for each month from a sales table containing columns: sale_id, product_id, sale_date, revenue. Use window functions.
Answer:
To solve this, use window functions to rank products within each month based on revenue. First, extract the month from sale_date using DATE_TRUNC('month', sale_date) or similar depending on the database (e.g., PostgreSQL or SQL Server).
The query would be:
WITH monthly_revenue AS (
SELECT
DATE_TRUNC('month', sale_date) AS month,
product_id,
SUM(revenue) AS total_revenue
FROM sales
GROUP BY month, product_id
),
ranked_products AS (
SELECT
month,
product_id,
total_revenue,
ROW_NUMBER() OVER (PARTITION BY month ORDER BY total_revenue DESC) AS rank
FROM monthly_revenue
)
SELECT
month,
product_id,
total_revenue
FROM ranked_products
WHERE rank <= 3
ORDER BY month, rank;This aggregates revenue per product per month, then ranks them using ROW_NUMBER(). The partition ensures ranking resets per month, and ordering by descending revenue gets the top performers. If ties are possible, consider DENSE_RANK() instead to handle equal rankings without skipping numbers.
Que 23. Explain how you would optimize a slow-running PySpark job that processes a large dataset with frequent joins and aggregations.
Answer:
Optimizing a PySpark job involves addressing bottlenecks in data shuffling, memory usage, and computation. First, analyze the job using Spark UI to identify stages with high shuffle read/write or GC time.
Use broadcast joins for small datasets: if one side is under 100MB, broadcast it with broadcast(df_small).join(df_large, "key") to avoid shuffling the large dataset. For data skew, where few keys dominate, apply salting: add a random suffix to keys like concat(key, lit("_"), rand() % 10) and adjust the join accordingly.
Tune partitioning: set spark.sql.shuffle.partitions to 2-3 times the number of cores, or use repartition on skewed columns. Cache datasets with df.cache() or persist(StorageLevel.MEMORY_AND_DISK) for reuse in multiple actions.
For aggregations, use groupBy with agg functions and avoid UDFs if possible—prefer built-in functions for catalyst optimization. Enable Adaptive Query Execution (AQE) in Spark 3+ with spark.sql.adaptive.enabled=true to dynamically coalesce partitions and handle skew.
Finally, monitor executor memory and increase it if spills occur, ensuring efficient file formats like Parquet for columnar reads.
Que 24. How do you implement Slowly Changing Dimension (SCD) Type 2 in a data warehouse using SQL or Spark, for a customer table where attributes like address can change over time?
Answer:
SCD Type 2 tracks historical changes by adding new rows for updates while keeping old versions, using effective dates and a surrogate key.
In SQL (e.g., in Snowflake or BigQuery), identify changes by comparing staging data with the dimension table. Use a merge statement:
MERGE INTO customer_dim AS target
USING customer_staging AS source
ON target.customer_id = source.customer_id AND target.is_current = TRUE
WHEN MATCHED AND (target.address <> source.address) THEN
UPDATE SET target.is_current = FALSE, target.end_date = CURRENT_DATE()
WHEN NOT MATCHED THEN
INSERT (surrogate_key, customer_id, address, start_date, end_date, is_current)
VALUES (NEXTVAL('surrogate_seq'), source.customer_id, source.address, CURRENT_DATE(), NULL, TRUE);After updating the matched row to expire it, insert a new row for the change.
In Spark, use DataFrames: join staging with current dimension on natural key, filter for changes, then union the unchanged rows, expired rows (with end_date set), and new versions (with start_date and is_current=True). Write to Delta Lake for ACID support: deltaTable.alias("target").merge(source.alias("source"), "target.customer_id = source.customer_id AND target.is_current = true")...
This preserves history for point-in-time queries.
Que 25. What is the difference between ETL and ELT, and in what scenarios would you choose one over the other for a data engineering project?
Answer:
ETL (Extract, Transform, Load) transforms data before loading into the target system, while ELT (Extract, Load, Transform) loads raw data first and transforms it in the target.
ETL is suitable for scenarios requiring strict data quality upfront, like compliance-heavy environments or when using legacy systems with limited compute. It reduces storage costs by cleaning data early but can be slower for large volumes due to intermediate processing.
ELT leverages the target’s compute power (e.g., BigQuery or Snowflake) for transformations, making it ideal for big data where raw ingestion is fast and schemas evolve. It’s more flexible for ad-hoc analysis but may increase storage costs with raw data.
For 2 years experience, prefer ELT in cloud setups for scalability, switching to ETL if transformations are complex and need custom logic before loading.
Que 26. Describe how you would handle data skew in a Spark join operation when one key dominates the dataset, causing uneven task execution.
Answer:
Data skew occurs when a few keys have disproportionately more data, leading to stragglers in Spark tasks. To mitigate, first detect it via Spark UI showing uneven task durations.
Apply salting: for the skewed side, explode the key by appending a random integer, e.g., df_skewed.withColumn("salted_key", concat(col("key"), lit("_"), floor(rand() * 10))). On the non-skewed side, replicate rows for each salt value using explode on an array of salts. Join on salted_key, then drop the salt.
Use broadcast join if one dataset fits in memory, avoiding shuffle altogether. Or, pre-aggregate the skewed keys separately and union with non-skewed results.
Enable AQE with spark.sql.adaptive.enabled=true and spark.sql.adaptive.skewJoin.enabled=true for automatic skew handling in Spark 3+. Repartition explicitly on the join key before joining to balance partitions.
Monitor with custom metrics if needed, ensuring even data distribution for efficient processing.
Que 27. Write a Python script using Pandas to clean a CSV file with missing values, duplicates, and inconsistent date formats, then output aggregated statistics.
Answer:
Start by loading the data with pd.read_csv('file.csv'). Handle missing values by filling numerical columns with mean (df['num_col'].fillna(df['num_col'].mean())) or dropping rows if critical (df.dropna(subset=['key_col'])).
Remove duplicates with df.drop_duplicates(subset=['id']). For dates, convert inconsistent formats using pd.to_datetime(df['date_col'], errors='coerce') and drop invalid ones.
Aggregate statistics like df.groupby('category')['sales'].agg(['sum', 'mean', 'count']).
Full script:
import pandas as pd
df = pd.read_csv('input.csv')
df['num_col'] = df['num_col'].fillna(df['num_col'].mean())
df = df.dropna(subset=['key_col'])
df = df.drop_duplicates(subset=['id'])
df['date_col'] = pd.to_datetime(df['date_col'], errors='coerce')
df = df.dropna(subset=['date_col']) # Drop invalid dates
stats = df.groupby('category')['sales'].agg(['sum', 'mean', 'count'])
stats.to_csv('output_stats.csv')This ensures clean data for further analysis.
Que 28. How would you design an Airflow DAG to orchestrate a daily batch job that extracts data from an API, transforms it in Spark, and loads it into a database?
Answer:
An Airflow DAG schedules tasks with dependencies. Define the DAG with DAG(dag_id='daily_batch', schedule_interval='@daily', start_date=datetime(2025, 1, 1)).
Tasks: Use PythonOperator for API extraction, calling a function to fetch data via requests and save to temporary storage like S3.
Next, SparkSubmitOperator for transformation: submit a PySpark job to process the data, e.g., filtering and aggregating.
Then, PostgresOperator or PythonOperator with SQLAlchemy to load into the database.
Set dependencies: extract >> transform >> load. Add sensors like ExternalTaskSensor if dependent on other DAGs, and use SLAs for monitoring.
For robustness, configure retries (retries=3, retry_delay=timedelta(minutes=5)) and email alerts on failure. This ensures reliable daily execution with logging.
Que 29. Explain a scenario where you would use caching versus persisting in Spark, and how to implement it for a DataFrame used in multiple computations.
Answer:
Caching stores a DataFrame in memory for fast access in repeated operations, while persisting allows specifying storage levels like disk or off-heap for spillover.
Use caching for in-memory speed when the DataFrame fits (e.g., small lookups), but persist with MEMORY_AND_DISK for larger ones to avoid recomputation on eviction.
Implement with df.cache() for default memory-only, or df.persist(StorageLevel.MEMORY_AND_DISK) for hybrid. Trigger computation with an action like df.count() to materialize.
Unpersist with df.unpersist() after use to free resources. This is practical in iterative ML pipelines or when joining the same DataFrame multiple times, reducing overall job time.
Que 30. How do you monitor and ensure data quality in a production data pipeline, including detecting anomalies like duplicates or schema drifts?
Answer:
Monitoring data quality involves validation at ingestion, transformation, and loading stages. Use tools like Great Expectations to define checks: expect no duplicates (expect_table_row_count_to_be_between), column values in range, or schema matches.
Integrate into the pipeline: in Airflow, add a task post-transformation to run validations, failing the DAG on issues. For anomalies, compute metrics like row counts, null percentages, or statistical summaries (e.g., mean/std dev) and alert via Slack if deviations exceed thresholds.
For schema drifts, use Deequ in Spark to profile data and compare against baselines. Log everything to ELK stack for querying, and set up dashboards in Grafana for visualizations.
Schedule periodic audits and reconcile source-to-target counts. This proactive approach prevents downstream issues in analytics.
Data Engineer Interview Questions for 7 Years Experience
Que. 31 How would you design a fault-tolerant data pipeline to process log data from multiple microservices and store aggregated metrics in a time-series database like InfluxDB?
Answer:
To design a fault-tolerant pipeline, start with log ingestion from microservices using a message queue like Kafka for durability. Each microservice publishes logs to a Kafka topic with a schema (e.g., JSON with timestamp, service_id, and metrics). Use multiple partitions for scalability.
For processing, use Apache Flink or Spark Streaming to consume logs. Configure checkpointing in Flink (env.enableCheckpointing(60000)) or Spark (writeStream.checkpointLocation) to ensure exactly-once processing. Aggregate metrics (e.g., requests per minute) using windowed operations, like Flink’s timeWindow(Time.minutes(1)) or Spark’s window("timestamp", "1 minute").
Write results to InfluxDB using its client library, batching writes to reduce overhead. Use retryOnFailure in the client for transient errors. Store raw logs in a data lake (e.g., S3 in Parquet) for auditing.
For fault tolerance, enable Kafka’s replication factor (e.g., 3) and use retries in processing jobs (spark.sql.shuffle.retries=3). Monitor with Prometheus for pipeline health and alert on failures via Slack. This ensures no data loss and reliable metric storage.
Que. 32 Write a SQL query to calculate the median order value per product category from an orders table with columns: order_id, product_category, order_value.
Answer:
To calculate the median, use a window function to assign row numbers and identify the middle value(s) per category.
WITH ranked_orders AS (
SELECT
product_category,
order_value,
ROW_NUMBER() OVER (PARTITION BY product_category ORDER BY order_value) AS rn,
COUNT(*) OVER (PARTITION BY product_category) AS cnt
FROM orders
),
median_values AS (
SELECT
product_category,
AVG(order_value) AS median_order_value
FROM ranked_orders
WHERE rn IN (FLOOR((cnt + 1)/2), CEIL((cnt + 1)/2))
GROUP BY product_category
)
SELECT
product_category,
median_order_value
FROM median_values
ORDER BY product_category;This assigns row numbers within each category, selects the middle row(s) based on count, and averages if even-numbered. For databases lacking window functions, use subqueries with LIMIT and OFFSET.
Que. 33 How do you optimize a Spark job that processes a 1TB dataset with frequent group-by operations causing high shuffle activity?
Answer:
High shuffle activity in group-by operations can be optimized by reducing data movement and optimizing resource allocation. Analyze the job via Spark UI to confirm shuffle bottlenecks (e.g., high spill or long stage times).
Reduce shuffle by increasing spark.sql.shuffle.partitions (e.g., 1000 for 1TB data) to balance tasks. Use repartition on the group-by key before aggregation to ensure even distribution and mitigate skew. For example, df.repartition("group_key").groupBy("group_key").agg(...).
Pre-aggregate smaller groups with a partial aggregation step using reduceByKey in RDDs or groupBy with approx_count_distinct for approximate results. Persist intermediate results with df.persist(StorageLevel.MEMORY_AND_DISK) to avoid recomputation.
Enable Adaptive Query Execution (spark.sql.adaptive.enabled=true) to dynamically adjust partitions. Use Parquet for columnar storage to reduce I/O. Monitor memory usage and increase spark.executor.memory if needed to minimize spills.
Que. 34 Write a Python script using Pandas to merge two CSV files, handle missing values, and export the result to a new CSV.
Answer:
Merge two CSVs, impute missing values, and export using Pandas.
import pandas as pd
df1 = pd.read_csv("customers.csv") # Columns: customer_id, name
df2 = pd.read_csv("orders.csv") # Columns: customer_id, order_amount
merged_df = df1.merge(df2, on="customer_id", how="left")
merged_df["order_amount"] = merged_df["order_amount"].fillna(0.0)
merged_df["name"] = merged_df["name"].fillna("Unknown")
merged_df.to_csv("merged_output.csv", index=False)This performs a left join, fills missing order_amount with 0.0 and name with “Unknown,” and exports without index. Validate input schemas before merging to ensure consistency.
Que. 35 How would you implement data partitioning in a data lake to improve query performance, and what factors influence partition key selection?
Answer:
Data partitioning in a data lake (e.g., S3) involves splitting data into folders based on columns like year, month, or region to enable partition pruning. Use Parquet or Delta Lake for efficient storage.
Choose partition keys based on query patterns: high-cardinality columns like date or customer_id work well if frequently filtered. For example, store data as s3://bucket/table/year=2025/month=08/. Avoid low-cardinality keys (e.g., status) to prevent large partitions.
Implement in Spark with df.write.partitionBy("year", "month").format("parquet").save("s3://bucket/table"). Use Delta Lake’s ZORDER on frequently joined columns to co-locate data.
Factors for key selection include query filters, data growth, and cardinality. Monitor partition sizes with AWS S3 Storage Lens and compact small files using OPTIMIZE in Delta Lake to avoid I/O overhead.
Que. 36 Write a SQL query to find customers who have placed orders in all product categories, using tables: orders (order_id, customer_id, product_category) and categories (category_id, category_name).
Answer:
Use a relational division approach to find customers present in all categories.
WITH customer_categories AS (
SELECT DISTINCT
o.customer_id,
o.product_category
FROM orders o
),
category_count AS (
SELECT COUNT(*) AS total_categories
FROM categories
),
customer_category_count AS (
SELECT
customer_id,
COUNT(*) AS cat_count
FROM customer_categories
GROUP BY customer_id
)
SELECT
ccc.customer_id
FROM customer_category_count ccc
JOIN category_count cc ON ccc.cat_count = cc.total_categories
ORDER BY ccc.customer_id;This counts unique categories per customer and matches against the total number of categories. The DISTINCT in customer_categories avoids double-counting.
Que. 37 How do you handle backfilling historical data in a data pipeline when a new transformation logic is introduced?
Answer:
Backfilling involves reprocessing historical data with new logic. Identify the date range and data sources (e.g., S3 or database). Use a batch processing framework like Spark or AWS Glue.
Create a temporary pipeline or DAG in Airflow to process historical data. For each partition (e.g., day), read data (spark.read.parquet("s3://bucket/year=2024/month=01/day=*")), apply the new transformation (e.g., updated aggregations), and write to a new or existing Delta Lake table with mergeInto for upserts.
Parallelize by partitioning on date and increasing Spark executors. Use spark.sql.shuffle.partitions to balance load. Test on a small subset first to validate logic.
Monitor progress with Airflow logs or Spark UI, and verify row counts match expectations. Once complete, switch the main pipeline to the new logic. This ensures consistency without disrupting live data.
Que. 38 Write a PySpark script to detect outliers in a dataset with columns: user_id, transaction_amount, transaction_date, using z-score.
Answer:
Use z-score to identify outliers based on standard deviation.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, mean, stddev
spark = SparkSession.builder.appName("OutlierDetection").getOrCreate()
df = spark.read.parquet("transactions")
stats = df.agg(
mean("transaction_amount").alias("mean_amt"),
stddev("transaction_amount").alias("std_amt")
).collect()[0]
mean_amt, std_amt = stats["mean_amt"], stats["std_amt"]
df_with_z = df.withColumn("z_score", (col("transaction_amount") - mean_amt) / std_amt)
outliers = df_with_z.filter((col("z_score") > 3) | (col("z_score") < -3))
outliers.write.format("parquet").save("s3://bucket/outliers")
spark.stop()This calculates the z-score for transaction_amount and flags records with |z_score| > 3 as outliers, saving them to a Parquet file.
Que. 39 Explain the difference between star schema and snowflake schema in data warehousing, and when would you choose one over the other?
Answer:
Star schema has a central fact table connected to denormalized dimension tables, forming a star-like structure. It’s simple, with fewer joins, optimizing query performance for analytical workloads. Snowflake schema normalizes dimensions into multiple related tables, reducing redundancy but increasing join complexity.
Choose star schema for simplicity and faster queries in BI tools like Tableau, especially with smaller datasets or when query performance is critical. It’s easier to maintain but uses more storage.
Choose snowflake schema for large datasets where storage efficiency matters, or when dimensions need hierarchical granularity (e.g., product -> subcategory -> category). It’s more complex but suits complex reporting needs.
For 2 years’ experience, prefer star schema for most analytics use cases due to its simplicity and performance.
Que. 40 How would you implement a real-time dashboard using data streamed from Kafka to display user activity metrics?
Answer:
To build a real-time dashboard, ingest user activity from Kafka using Spark Structured Streaming. Define a schema for events (e.g., user_id, action, timestamp) and read the stream with spark.readStream.format("kafka").
Aggregate metrics like active users per minute using groupBy(window("timestamp", "1 minute"), "action").count(). Write results to an in-memory table (writeStream.outputMode("complete").format("memory")) for querying.
Use a dashboard tool like Grafana with a plugin (e.g., Spark SQL or Redis sink) to query the in-memory table. Alternatively, write to Redis or Elasticsearch for low-latency access and connect Grafana to visualize metrics.
Ensure fault tolerance with Kafka replication and Spark checkpointing. Monitor latency with Prometheus and scale Spark executors for high throughput. This delivers real-time insights with minimal delay.
Data Engineer Interview Questions for 10 Years Experience
Que. 41 Write a SQL query to find the top 3 most frequently purchased product pairs in a given month from an orders table with columns: order_id, product_id, order_date.
Answer:
Identify product pairs within the same order and count their occurrences.
WITH order_pairs AS (
SELECT
o1.product_id AS product1,
o2.product_id AS product2,
COUNT(*) AS pair_count
FROM orders o1
JOIN orders o2 ON o1.order_id = o2.order_id AND o1.product_id < o2.product_id
WHERE DATE_TRUNC('month', o1.order_date) = '2025-08-01'
GROUP BY o1.product_id, o2.product_id
)
SELECT
product1,
product2,
pair_count
FROM order_pairs
ORDER BY pair_count DESC
LIMIT 3;The self-join ensures pairs are unique (product_id < product_id), and the WHERE filters for August 2025. Results show the top 3 pairs by frequency.
Que. 42 How do you ensure GDPR compliance in a data pipeline handling EU customer data?
Answer:
GDPR compliance requires data protection, consent, and auditability. Encrypt PII in transit (Kafka SSL) and at rest (S3 SSE-KMS). Anonymize or pseudonymize sensitive fields (e.g., hash emails in Spark: sha2(col("email"), 256)).
Implement access controls with IAM roles and Lake Formation, granting least privilege. Use a consent management system to track user permissions, filtering out non-consented data in ETL jobs (WHERE consent = true).
Support data deletion requests with soft deletes in Delta Lake (UPDATE table SET is_deleted = true) and purge after retention (e.g., VACUUM after 30 days). Log access and transformations in CloudTrail for audits.
Tag PII in the data catalog and monitor with AWS Macie. Train teams on GDPR and document processes. This ensures compliance while processing EU data.
Que. 43 Write a Python script using boto3 to list all objects in an S3 bucket and filter for files modified in the last 24 hours.
Answer:
Use boto3 to list and filter objects by modification time.
import boto3
from datetime import datetime, timedelta, timezone
s3_client = boto3.client("s3")
bucket_name = "my-data-lake"
cutoff_time = datetime.now(timezone.utc) - timedelta(hours=24)
try:
response = s3_client.list_objects_v2(Bucket=bucket_name)
recent_objects = [
obj["Key"] for obj in response.get("Contents", [])
if obj["LastModified"] >= cutoff_time
]
print(f"Objects modified in last 24 hours: {recent_objects}")
except ClientError as e:
print(f"Error listing objects: {e}")This lists all objects and filters those modified within 24 hours using LastModified. Handle pagination with ContinuationToken for large buckets.
Que. 44 How do you handle data lineage tracking in a complex data pipeline spanning multiple systems?
Answer:
Data lineage tracks data flow from source to destination. Use a tool like Apache Atlas or OpenLineage to capture metadata across systems (Kafka, Spark, Snowflake).
Ingest lineage by integrating with pipeline tools: Spark’s OpenLineage connector logs transformations, while Airflow’s OpenLineage plugin captures task dependencies. For databases, use query logs or tools like dbt for lineage graphs.
Store lineage in a catalog with metadata like source, transformations, and destination (e.g., s3://bucket/table). Visualize in Atlas UI or Grafana for end-to-end tracking.
Validate lineage by auditing transformations against expected flows. Automate metadata collection with hooks in ETL jobs and monitor for gaps. This ensures traceability for debugging and compliance.
Que. 45 Write a PySpark script to calculate the percentage of total sales contributed by each region from a sales table.
Answer:
Compute regional sales and divide by total sales.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum
spark = SparkSession.builder.appName("SalesPercentage").getOrCreate()
df = spark.read.parquet("s3://bucket/sales")
total_sales = df.agg(sum("sales_amount").alias("total")).collect()[0]["total"]
region_sales = df.groupBy("region").agg(sum("sales_amount").alias("region_total"))
result = region_sales.withColumn("percentage", (col("region_total") / total_sales) * 100)
result.write.format("parquet").save("s3://bucket/region_percentages")
spark.stop()This groups by region, sums sales_amount, calculates percentages, and saves results. Use coalesce(1) for single-file output if needed.
Que. 46 How would you migrate a data pipeline from on-premises Hadoop to a cloud-based solution like AWS EMR?
Answer:
Migrating a Hadoop pipeline to AWS EMR involves planning, data transfer, and pipeline redesign. Assess the existing pipeline: identify jobs (e.g., MapReduce, Hive), data sources (HDFS), and dependencies.
Transfer data to S3 using distcp (hadoop distcp hdfs://source s3://bucket). Convert HDFS data to Parquet for efficiency. Refactor MapReduce jobs to Spark or Hive on EMR, leveraging EMR’s managed Hadoop environment.
Update orchestration: replace on-premises schedulers with Airflow on AWS MWAA, rewriting DAGs to use EmrAddStepsOperator. Test jobs on a small EMR cluster, validating output against Hadoop results.
Configure EMR with auto-scaling for cost efficiency and enable S3 encryption. Monitor with CloudWatch and migrate incrementally by data source to minimize disruption. Document changes and train teams on AWS tools.
Que. 47 Write a SQL query to find the average time between consecutive orders for each customer from an orders table (order_id, customer_id, order_date).
Answer:
Calculate time differences between consecutive orders using LAG.
WITH order_diffs AS (
SELECT
customer_id,
order_date,
LAG(order_date) OVER (PARTITION BY customer_id ORDER BY order_date) AS prev_order_date,
DATEDIFF('day', LAG(order_date) OVER (PARTITION BY customer_id ORDER BY order_date), order_date) AS days_diff
FROM orders
)
SELECT
customer_id,
AVG(days_diff) AS avg_days_between_orders
FROM order_diffs
WHERE days_diff IS NOT NULL
GROUP BY customer_id
ORDER BY customer_id;This uses LAG to get the previous order date, computes the difference in days, and averages per customer. Adjust DATEDIFF syntax for your database (e.g., DATEDIFF(day, ...) for SQL Server).
Que. 48 How do you tune Apache Kafka for high-throughput ingestion of event data?
Answer:
Tuning Kafka for high throughput involves optimizing producers, brokers, and consumers. For producers, set batch.size=16384 and linger.ms=5 to batch messages, increasing compression (compression.type=snappy). Use acks=1 for a balance of reliability and speed.
On brokers, increase num.io.threads and num.network.threads (e.g., 8 each) to handle more connections. Set num.partitions (e.g., 50) per topic for parallelism, and ensure replication.factor=3 for fault tolerance. Allocate sufficient memory and use SSDs for log.dirs.
For consumers, increase fetch.max.bytes (e.g., 50MB) and use multiple consumer groups for parallel processing. Tune max.partition.fetch.bytes to control data per partition.
Monitor with Prometheus and JMX metrics for lag and throughput. Test with tools like kafka-producer-perf-test to validate performance. This maximizes ingestion capacity.
Que. 49 Write a Python script to validate JSON data against a schema before loading into a database.
Answer:
Use jsonschema to validate JSON data.
import json
from jsonschema import validate, ValidationError
schema = {
"type": "object",
"properties": {
"user_id": {"type": "integer"},
"name": {"type": "string"},
"amount": {"type": "number"}
},
"required": ["user_id", "name", "amount"]
}
with open("data.json") as f:
data = json.load(f)
try:
if isinstance(data, list):
for item in data:
validate(instance=item, schema=schema)
else:
validate(instance=data, schema=schema)
print("JSON validation passed")
# Proceed to load into database
except ValidationError as e:
print(f"Validation failed: {e}")This validates JSON against a schema requiring user_id, name, and amount. Extend with database insertion (e.g., SQLAlchemy) after validation.
Que. 50 How do you implement data quality checks in a pipeline using Great Expectations?
Answer:
Great Expectations validates data against defined expectations. Install great_expectations and initialize a project (great_expectations init).
Define expectations in a suite: e.g., expect_column_values_to_not_be_null("user_id"), expect_column_values_to_be_between("amount", min_value=0). Create a checkpoint to run validations:
import great_expectations as ge
df = ge.read_csv("data.csv")
expectation_suite = df.expectation_suite()
df.expect_column_values_to_not_be_null("user_id")
df.expect_column_values_to_be_between("amount", min_value=0)
df.validate(expectation_suite=expectation_suite, run_name="data_validation")Integrate into Airflow with a PythonOperator to run the checkpoint after ETL steps. Store results in S3 or a database for auditing. Set alerts via Slack for failures.
Use a data catalog to track validated datasets and schedule periodic checks. This ensures consistent data quality across the pipeline.
Conclusion
We have already shared the essential questions for Data Engineer Interview Questions for Experienced professionals. This comprehensive Data Engineer Guide includes interview questions for experienced candidates with advanced industry experience, covering complex architectural scenarios and leadership challenges that employers evaluate. The data engineering industry is rapidly evolving with Kubernetes orchestration, serverless computing, and real-time analytics becoming standard requirements for senior roles.
These Data Engineer Interview Questions for Experienced professionals provide the strategic foundation needed to advance your career, covering distributed computing to infrastructure automation. With proper preparation using these Data Engineer Interview Questions for Experienced and understanding current industry demands, you’ll be well-positioned to secure senior data engineering positions.
