Tiger Analytics: Comprehensive Technical Interview Revision Guide

About Tiger Analytics

Tiger Analytics is a leading global analytics consulting and data engineering firm specializing in digital transformation, business intelligence, and advanced analytics. The company works with Fortune 500 enterprises across industries to build scalable data platforms and derive actionable insights from complex datasets. Tiger Analytics is known for its expertise in cloud technologies (especially Azure), big data processing frameworks like Apache Spark, and end-to-end data pipeline development. Their technical teams typically work with modern data stacks including Databricks, Azure Data Lake, Azure Synapse Analytics, and Apache Spark.

---

PySpark Fundamentals

1. Explain Lazy Evaluation in PySpark

Answer:

Lazy evaluation is one of the most important concepts in PySpark that distinguishes it from traditional Python operations. When you perform transformations on RDDs or DataFrames, PySpark doesn’t immediately execute them. Instead, it builds a Directed Acyclic Graph (DAG) of transformations.

Here’s what happens:

• Transformations (like map(), filter(), select(), join()) are recorded as a plan but not executed.
• Actions (like collect(), show(), count(), write()) trigger the actual computation.
• The optimizer then analyzes the entire DAG and optimizes it before execution, a process called Catalyst optimization.

Why does this matter?

1. Performance: PySpark can optimize the entire chain of operations together rather than executing them one by one.
2. Resource Efficiency: Unnecessary operations can be eliminated during optimization.
3. Pipeline Resilience: If a task fails mid-way, Spark can recalculate from any checkpoint using the DAG.

# These lines don't execute yet (transformations are lazy)
df = spark.read.csv("data.csv", header=True)
filtered_df = df.filter(df.age > 25)
selected_df = filtered_df.select("name", "age")

# This line triggers execution (action)
selected_df.show()  # NOW Spark builds the DAG and executes

---

2. How Does Caching Work in PySpark?

Answer:

Caching (also called persistence) stores the intermediate results of a DataFrame or RDD in memory so that when you perform multiple actions on the same data, you don’t need to recompute it from the source.

How it works:

When you call .cache() or .persist() on a DataFrame, PySpark marks it for caching. The actual data is stored in memory the first time an action is performed. Subsequent actions on that cached DataFrame will reuse the in-memory data.

Different Cache Levels:

from pyspark import StorageLevel

df = spark.read.csv("large_file.csv")

# MEMORY_ONLY (default)
df.cache()

# More explicit control
df.persist(StorageLevel.MEMORY_ONLY)           # RAM only
df.persist(StorageLevel.MEMORY_AND_DISK)       # RAM + Disk spillover
df.persist(StorageLevel.DISK_ONLY)             # Disk only
df.persist(StorageLevel.MEMORY_ONLY_2)         # RAM with 2x replication

When to use caching:

• When the same DataFrame is used multiple times in your pipeline
• When a transformation is computationally expensive
• When performing exploratory data analysis with multiple queries

Important Note: Always remove cache after use to free up memory:

df.unpersist()

---

3. What is the Difference Between Wide and Narrow Transformations?

Answer:

This distinction is crucial for understanding Spark’s execution model and partitioning strategy.

Narrow Transformations:

• Each partition in the input DataFrame produces a single partition in the output.
• Data doesn’t need to move between partitions.
• No shuffle operation is required.
• Fast and memory-efficient.

Examples: map(), filter(), select(), withColumn(), drop()

# All narrow transformations
df.filter(df.salary > 50000)      # filters within each partition
df.select("name", "age")          # selects columns within each partition
df.withColumn("bonus", df.salary * 0.1)  # calculation within each partition

Wide Transformations:

• Input partitions contribute data to multiple output partitions.
• Requires shuffling data across the network.
• Expensive in terms of computation and I/O.
• Creates a barrier between stages in the execution plan.

Examples: join(), groupBy(), repartition(), distinct(), orderBy()

# All wide transformations
df1.join(df2, "id")               # data shuffles across network
df.groupBy("department").sum()    # groups shuffle data by key
df.repartition(100)               # explicitly redistributes data
df.orderBy("salary")              # requires full shuffle to sort

Visual Difference:

Narrow:  Partition 1 → Partition 1
         Partition 2 → Partition 2
         Partition 3 → Partition 3

Wide:    Partition 1 \
         Partition 2  → Partition 1
         Partition 3 / → Partition 2
                      → Partition 3

---

10. Explain How Broadcast Joins Improve Performance in PySpark

Answer:

A broadcast join is an optimization technique where one DataFrame (typically the smaller one) is broadcast to all executor nodes. The larger DataFrame is then joined with this broadcast copy, eliminating the need for a shuffle operation.

How it works:

1. The smaller DataFrame is collected to the driver node.
2. The entire DataFrame is serialized and sent to all executor nodes.
3. Each executor performs a local hash join with its partition of the larger DataFrame.
4. No network shuffle is required.

Performance Benefits:

• Eliminates shuffle overhead for the smaller DataFrame
• Can reduce execution time by 50-70% for large datasets
• Reduces network I/O significantly

Example:

from pyspark.sql.functions import broadcast

# Without broadcast (triggers shuffle)
result = large_df.join(small_df, "id")

# With broadcast (much faster)
result = large_df.join(broadcast(small_df), "id")

When to use broadcast joins:

• One DataFrame is significantly smaller than the other
• The smaller DataFrame can fit in memory (typically < 100 MB)
• You want to avoid shuffle operations

Important Consideration:

Broadcasting is automatic for small DataFrames (< 10 MB by default), but you can configure it:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 104857600)  # 100 MB

---

11. Describe the Role of the Driver and Executor in Spark Architecture

Answer:

Understanding the driver-executor architecture is fundamental to optimizing Spark applications.

The Driver:

The driver is the central control point of a Spark application. It:

• Maintains the state of the Spark Session
• Receives user code and translates it into tasks
• Schedules tasks on executors
• Monitors executor progress
• Collects results from executors and returns them to the user
• Handles caching metadata
• Runs the DAG scheduler and task scheduler

The Executors:

Executors are worker processes that:

• Execute the actual tasks assigned by the driver
• Perform computations and return results to the driver
• Store cached data in their memory
• Communicate task status back to the driver
• Run code in parallel

Communication Flow:

User Code (Driver)
    ↓
DAG Creation
    ↓
Task Scheduling (Driver)
    ↓
Task Execution (Executors) → Results sent back to Driver
    ↓
Result Collection

Example Architecture:

┌─────────────────────┐
│   Driver Node       │
│ - SparkContext      │
│ - DAG Scheduler     │
│ - Task Scheduler    │
└──────────┬──────────┘
           │
    ┌──────┼──────┐
    ↓      ↓      ↓
┌────┐  ┌────┐  ┌────┐
│ Ex │  │ Ex │  │ Ex │  (Multiple Executors)
│ 1  │  │ 2  │  │ 3  │
└────┘  └────┘  └────┘

Key Insights:

• The driver is the bottleneck for collecting results; avoid using collect() on large datasets
• Each executor has its own JVM and cache storage
• The number of executors and cores directly impacts parallelism

---

19. Explain Adaptive Query Execution (AQE) in Spark

Answer:

Adaptive Query Execution (AQE) is an optimization feature in Spark SQL that reoptimizes queries during runtime based on actual statistics collected from completed stages. It represents a shift from static optimization to dynamic optimization.

How AQE Works:

1. Spark executes stages and collects runtime statistics (actual data size, skewness, etc.)
2. Between stages, Spark reoptimizes the query plan using these statistics
3. The execution plan is adjusted for remaining stages

Three Main Optimizations in AQE:

1. Coalesce Partitions

If a stage produces significantly fewer partitions than expected, AQE automatically reduces the number of partitions to eliminate unnecessary parallelism overhead.

# Enable AQE (Spark 3.0+, enabled by default)
spark.conf.set("spark.sql.adaptive.enabled", "true")

2. Convert Sort-Merge Join to Broadcast Join

If during execution a join’s smaller table turns out to be much smaller than anticipated, AQE converts it to a broadcast join.

3. Skew Join Optimization

Handles scenarios where data is skewed (some partitions have much more data than others), which can create stragglers. AQE splits skewed partitions and duplicates the other side.

Example Scenario:

# In this query, Spark initially plans sort-merge join
# But if at runtime the small_df is tiny, it becomes a broadcast join
result = large_df.join(small_df, "id")

# AQE automatically optimizes this without code changes
result.show()

Benefits:

• Adapts to real data distribution
• Reduces execution time for queries with uncertain statistics
• Eliminates manual partitioning hints
• Handles data skew automatically

---

22. Write PySpark Code to Perform a Left Join Between Two DataFrames

Answer:

from pyspark.sql import SparkSession

# Initialize Spark Session
spark = SparkSession.builder.appName("JoinExample").getOrCreate()

# Create sample DataFrames
employees = spark.createDataFrame([
    (1, "Alice", "Engineering"),
    (2, "Bob", "Sales"),
    (3, "Charlie", "Engineering"),
    (4, "David", "HR")
], ["emp_id", "name", "department"])

projects = spark.createDataFrame([
    (1, "Project A"),
    (1, "Project B"),
    (2, "Project C"),
    (5, "Project D")  # No matching employee
], ["emp_id", "project"])

# Perform Left Join
# Left DataFrame (employees) retains all rows, matches from projects where possible
result = employees.join(projects, "emp_id", "left")

result.show()

Output:

+------+-------+-------------+---------+
|emp_id|   name|   department|  project|
+------+-------+-------------+---------+
|     1|  Alice|   Engineering|Project A|
|     1|  Alice|   Engineering|Project B|
|     2|    Bob|        Sales|Project C|
|     3|Charlie|   Engineering|     null|
|     4|  David|            HR|     null|
+------+-------+-------------+---------+

Alternative Syntax:

# More explicit join syntax
result = employees.join(projects, employees.emp_id == projects.emp_id, "left")

# Renaming to avoid column conflicts (if both have emp_id with different meanings)
result = employees.alias("e").join(
    projects.alias("p"), 
    employees.emp_id == projects.emp_id, 
    "left"
).select("e.emp_id", "e.name", "e.department", "p.project")

---

9. Write PySpark Code to Calculate the Total Sales for Each Product Category

Answer:

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum as spark_sum, col, round as spark_round

# Initialize Spark Session
spark = SparkSession.builder.appName("SalesAnalysis").getOrCreate()

# Create sample sales data
sales_data = spark.createDataFrame([
    ("Electronics", "Laptop", 1200.00),
    ("Electronics", "Mouse", 25.50),
    ("Electronics", "Keyboard", 85.00),
    ("Clothing", "T-Shirt", 15.99),
    ("Clothing", "Jeans", 49.99),
    ("Home", "Pillow", 35.00),
    ("Home", "Bedsheet", 59.99),
    ("Electronics", "Monitor", 350.00)
], ["category", "product", "price"])

# Method 1: Using groupBy and sum
total_sales_by_category = sales_data.groupBy("category").agg(
    spark_sum("price").alias("total_sales")
).orderBy(col("total_sales").desc())

total_sales_by_category.show()

# Method 2: With rounding for currency precision
total_sales_by_category = sales_data.groupBy("category").agg(
    spark_round(spark_sum("price"), 2).alias("total_sales")
).orderBy(col("total_sales").desc())

total_sales_by_category.show()

# Method 3: Multiple aggregations
detailed_analysis = sales_data.groupBy("category").agg(
    spark_sum("price").alias("total_sales"),
    spark_round(spark_sum("price") / count("*"), 2).alias("average_price"),
    count("*").alias("product_count")
)

detailed_analysis.show()

Output:

+-------------+------------+
|     category|  total_sales|
+-------------+------------+
|   Electronics|     1660.50|
|         Home|       94.99|
|    Clothing|       65.98|
+-------------+------------+

---

SQL Optimization

7. Write a SQL Query to Find the nth Highest Salary in a Table

Answer:

-- Method 1: Using ROW_NUMBER (Most Common)
SELECT salary
FROM (
    SELECT 
        salary,
        ROW_NUMBER() OVER (ORDER BY salary DESC) as rank
    FROM employees
)
WHERE rank = 5;  -- For 5th highest salary

-- Method 2: Using DENSE_RANK (Handles ties)
-- Use this if you want consecutive ranking despite duplicates
SELECT salary
FROM (
    SELECT 
        salary,
        DENSE_RANK() OVER (ORDER BY salary DESC) as rank
    FROM employees
)
WHERE rank = 5;

-- Method 3: Using OFFSET and LIMIT (PostgreSQL, SQL Server 2012+)
SELECT DISTINCT salary
FROM employees
ORDER BY salary DESC
OFFSET 4 ROWS     -- Skip first (n-1) rows
FETCH NEXT 1 ROW ONLY;

-- Method 4: Using subquery (Works on most databases)
SELECT MAX(salary) as nth_highest_salary
FROM employees e1
WHERE (
    SELECT COUNT(DISTINCT e2.salary)
    FROM employees e2
    WHERE e2.salary >= e1.salary
) = 5;  -- For 5th highest

-- Method 5: Create a dynamic query with parameter
DECLARE @n INT = 5;
SELECT salary
FROM (
    SELECT 
        salary,
        ROW_NUMBER() OVER (ORDER BY salary DESC) as rank
    FROM employees
)
WHERE rank = @n;

Key Differences:

Method	Handles Ties	Database Support	Simplicity
ROW_NUMBER	No	All modern DBs	Very High
DENSE_RANK	Yes	All modern DBs	High
OFFSET/LIMIT	No	PostgreSQL, SQL Server, MySQL	Very High
Subquery	Yes	All databases	Medium

Complete Example with Department Filter:

-- Find 3rd highest salary in each department
SELECT 
    department,
    salary,
    employee_name
FROM (
    SELECT 
        department,
        salary,
        employee_name,
        ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) as rank
    FROM employees
)
WHERE rank = 3;

---

13. Write a SQL Query to Find Employees with Salaries Greater than Department Average

Answer:

-- Method 1: Using CTE (Clearest and Most Readable)
WITH dept_avg AS (
    SELECT 
        department,
        AVG(salary) as avg_salary
    FROM employees
    GROUP BY department
)
SELECT 
    e.employee_id,
    e.employee_name,
    e.department,
    e.salary,
    da.avg_salary,
    ROUND(e.salary - da.avg_salary, 2) as salary_above_avg
FROM employees e
INNER JOIN dept_avg da ON e.department = da.department
WHERE e.salary > da.avg_salary
ORDER BY e.department, e.salary DESC;

-- Method 2: Using Subquery
SELECT 
    employee_id,
    employee_name,
    department,
    salary
FROM employees e1
WHERE salary > (
    SELECT AVG(salary)
    FROM employees e2
    WHERE e2.department = e1.department
)
ORDER BY department, salary DESC;

-- Method 3: Using Window Function (SQL Server, PostgreSQL, Oracle)
SELECT 
    employee_id,
    employee_name,
    department,
    salary,
    AVG(salary) OVER (PARTITION BY department) as dept_avg_salary
FROM employees
WHERE salary > AVG(salary) OVER (PARTITION BY department)
ORDER BY department, salary DESC;

Sample Output:

employee_id  employee_name  department    salary  avg_salary  salary_above_avg
    1        Alice          Engineering   95000   70000       25000
    2        Bob            Engineering   85000   70000       15000
    5        Charlie        Sales         72000   55000       17000
    8        Diana          HR            62000   58000       4000

Key Points:

• Method 1 (CTE) is most readable and maintainable
• Method 2 (Subquery) works in all SQL databases
• Method 3 (Window Functions) is most efficient for large datasets in modern databases
• Always use INNER JOIN in method 1 to avoid NULL comparisons

---

Azure Data Platform

4. How do you Optimize Query Performance in Azure SQL Database?

Answer:

Query performance optimization in Azure SQL Database involves multiple strategies:

1. Query Plan Analysis

Always examine execution plans to identify bottlenecks:

-- Enable statistics collection
SET STATISTICS IO ON;
SET STATISTICS TIME ON;

-- Your query here
SELECT * FROM large_table WHERE status = 'active';

-- Results show I/O and CPU time
SET STATISTICS IO OFF;
SET STATISTICS TIME OFF;

2. Indexing Strategy

-- Clustered index on frequently searched columns
CREATE CLUSTERED INDEX idx_customers_id ON customers(customer_id);

-- Non-clustered indexes for WHERE and JOIN conditions
CREATE NONCLUSTERED INDEX idx_customers_email 
ON customers(email) INCLUDE (customer_name);

-- Check existing indexes
SELECT * FROM sys.indexes WHERE object_id = OBJECT_ID('customers');

3. Query Optimization Techniques

-- Use SELECT with specific columns (not SELECT *)
SELECT customer_id, email, name  -- Good
FROM customers
WHERE status = 'active';

-- Avoid functions in WHERE clauses
SELECT * FROM orders 
WHERE YEAR(order_date) = 2024;  -- Bad: prevents index use

SELECT * FROM orders 
WHERE order_date >= '2024-01-01' 
AND order_date < '2025-01-01';  -- Good

-- Use NOLOCK for read-only queries to reduce blocking
SELECT * FROM customers WITH (NOLOCK)
WHERE status = 'active';

4. Materialized Views for Aggregations

-- Create materialized view for expensive aggregations
CREATE MATERIALIZED VIEW vw_sales_summary
AS
SELECT 
    product_id,
    YEAR(order_date) as year,
    SUM(quantity) as total_quantity,
    SUM(amount) as total_amount
FROM orders
GROUP BY product_id, YEAR(order_date);

-- Refresh when source data changes
ALTER MATERIALIZED VIEW vw_sales_summary REBUILD;

5. Partitioning for Large Tables

-- Partition table by date range for better performance
CREATE TABLE orders_partitioned (
    order_id INT,
    customer_id INT,
    order_date DATE,
    amount DECIMAL(10, 2)
) WITH (DISTRIBUTION = HASH(customer_id), CLUSTERED COLUMNSTORE INDEX)
PARTITION (order_date RANGE LEFT FOR VALUES 
    ('2023-01-01', '2023-07-01', '2024-01-01', '2024-07-01')
);

6. Query Hints and Batch Mode

-- Use query hints for specific optimization
SELECT * FROM orders
WHERE customer_id = 123
OPTION (RECOMPILE);  -- Recompile to get fresh statistics

-- In Azure SQL DW, use OPTION (BROADCAST) for small tables
SELECT * FROM large_table l
INNER JOIN small_table s ON l.id = s.id
OPTION (BROADCAST(small_table));

7. Configuration and Monitoring

-- Check slow queries
SELECT query_id, execution_type_desc, total_cpu_time, total_elapsed_time
FROM sys.query_store_runtime_stats
ORDER BY total_cpu_time DESC;

-- Get index fragmentation
SELECT object_name(ips.object_id) as table_name,
       i.name as index_name,
       ips.avg_fragmentation_in_percent
FROM sys.dm_db_index_physical_stats(DB_ID(), NULL, NULL, NULL, 'LIMITED') ips
INNER JOIN sys.indexes i ON ips.object_id = i.object_id 
                         AND ips.index_id = i.index_id
WHERE ips.avg_fragmentation_in_percent > 10;

-- Rebuild fragmented indexes
ALTER INDEX ALL ON table_name REBUILD;

---

5. Describe the Process of Integrating ADF with Azure Synapse Analytics

Answer:

Integration between Azure Data Factory (ADF) and Azure Synapse Analytics creates a powerful end-to-end analytics platform.

Architecture Overview:

Data Sources → ADF Pipeline → Data Lake/Staging → Synapse Analytics → BI Tools
                (Orchestration)   (Transformation)   (Analytics)

Step-by-Step Integration:

1. Set Up Connections

Create linked services in ADF pointing to Synapse:

Data Factory → Linked Services → New Linked Service
Select: Azure Synapse Analytics
Connection Details:
- Account name: yoursynapse.sql.azuresynapseanalytics.net
- Authentication: Managed Identity (recommended)

2. Create Synapse Linked Service

In ADF, create a linked service for Synapse:

{
    "name": "AzureSynapseAnalyticsLinkedService",
    "properties": {
        "type": "AzureSqlDW",
        "typeProperties": {
            "connectionString": "Server=tcp:yourserver.sql.azuresynapseanalytics.net,1433;Initial Catalog=yourdb;Persist Security Info=False;User ID=sqladmin;Password=YourPassword;MultipleActiveResultSets=False;Encrypt=True;TrustServerCertificate=False;Connection Timeout=30;"
        }
    }
}

3. Copy Data Activity in Pipeline

{
    "name": "CopyDataToSynapse",
    "type": "Copy",
    "inputs": [
        {
            "referenceName": "SourceBlobDataset",
            "type": "DatasetReference"
        }
    ],
    "outputs": [
        {
            "referenceName": "SynapseDataset",
            "type": "DatasetReference"
        }
    ],
    "typeProperties": {
        "source": {
            "type": "DelimitedTextSource"
        },
        "sink": {
            "type": "SqlDWSink",
            "preCopyScript": "TRUNCATE TABLE [dbo].[staging_table]",
            "sqlWriterTableType": "SynapseTableType"
        }
    }
}

4. Use Synapse Pipelines for Advanced Processing

Execute Spark notebooks or SQL scripts in Synapse directly from ADF:

{
    "name": "ExecuteSynapseNotebook",
    "type": "SynapseNotebook",
    "typeProperties": {
        "notebook": {
            "referenceName": "TransformDataNotebook",
            "type": "NotebookReference"
        },
        "parameters": {
            "input_path": "/data/raw",
            "output_path": "/data/processed"
        }
    }
}

5. Error Handling and Retry Logic

{
    "retryPolicy": {
        "count": 3,
        "intervalInSeconds": 30
    },
    "onFailure": [
        {
            "type": "SetVariable",
            "typeProperties": {
                "variableName": "ErrorFlag",
                "value": "true"
            }
        }
    ]
}

Key Integration Points:

Component	Purpose
ADF Pipeline	Orchestrates data movement and transformations
Synapse Linked Service	Connects ADF to Synapse Analytics
Copy Activity	Moves data from sources to Synapse
Synapse Notebooks	Execute Spark transformations
SQL Scripts	Run T-SQL for data processing
Synapse Pipelines	Alternative orchestration within Synapse

---

6. How do you Handle Schema Evolution in Azure Data Lake?

Answer:

Schema evolution is the ability to add or modify columns without breaking existing pipelines.

Strategy 1: Allow Schema Drift in ADF

Configure copy activity to handle schema changes:

{
    "name": "CopyWithSchemaDrift",
    "type": "Copy",
    "inputs": [
        {
            "referenceName": "FlexibleSourceDataset",
            "type": "DatasetReference"
        }
    ],
    "typeProperties": {
        "source": {
            "type": "JsonSource"
        },
        "sink": {
            "type": "ParquetSink",
            "schemaMapping": []
        },
        "schemaMapping": [
            {
                "source": {
                    "path": "$['existing_column']"
                },
                "sink": {
                    "name": "existing_column"
                }
            }
        ],
        "enableSchemaDrift": true  // Allow new columns
    }
}

Strategy 2: Use Delta Lake for Automatic Evolution

Delta Lake automatically handles schema changes:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SchemaEvolution").getOrCreate()

# Read existing data
df_v1 = spark.read.delta("/mnt/datalake/customer_data")

# Add new columns in the DataFrame
df_v2 = df_v1.withColumn("new_address", lit(None))
                .withColumn("new_phone", lit(None))

# Write with schema evolution enabled (Delta default behavior)
df_v2.write.format("delta").mode("overwrite").save("/mnt/datalake/customer_data")

Strategy 3: Explicit Schema Versioning

Maintain schema versions and validate during ingestion:

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Define versioned schemas
schema_v1 = StructType([
    StructField("customer_id", IntegerType()),
    StructField("name", StringType())
])

schema_v2 = StructType([
    StructField("customer_id", IntegerType()),
    StructField("name", StringType()),
    StructField("email", StringType())  # New in v2
])

# Read with appropriate schema
df = spark.read.schema(schema_v2).csv("/path/to/file.csv")

# Validate and transform
def normalize_schema(df, target_version):
    if target_version == "v2":
        if "email" not in df.columns:
            from pyspark.sql.functions import lit
            df = df.withColumn("email", lit(None))
    return df

normalized_df = normalize_schema(df, "v2")

Strategy 4: Column Pruning and Safe Schema Updates

# Archive old data with schema metadata
def write_with_schema_metadata(df, path, version):
    df.write \
        .mode("overwrite") \
        .option("schemaVersion", version) \
        .parquet(f"{path}/v{version}")
    
    # Also save schema separately
    schema_json = df.schema.json()
    with open(f"{path}/v{version}_schema.json", 'w') as f:
        f.write(schema_json)

write_with_schema_metadata(df, "/mnt/datalake/data", "2.0")

Strategy 5: Auto-Merge for Column Addition

# When a new column appears in source
from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/mnt/datalake/customer_data")

# Merge new data (auto-adds columns)
try:
    delta_table.merge(
        new_df, 
        "existing_id = new_id"
    ).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
except Exception as e:
    if "column not found" in str(e):
        # Auto-evolve schema by rewriting
        combined = existing_df.union(new_df.select(existing_df.columns))
        combined.write.format("delta").mode("overwrite").save("/mnt/datalake/customer_data")

Best Practices:

• Use Delta Lake for automatic schema evolution capabilities
• Monitor schema changes with metadata tables
• Test schema changes in development before production
• Document schema versions and maintain backward compatibility
• Use ADLS Gen2 hierarchical structure to organize versioned data

---

8. How do you Implement CI/CD Pipelines for Deploying ADF and Databricks Solutions?

Answer:

CI/CD for data engineering requires orchestrating code deployment, testing, and infrastructure management.

1. Source Control Strategy

Repository Structure:
├── adf/
│   ├── pipelines/
│   ├── datasets/
│   ├── linkedServices/
│   └── arm_template.json
├── databricks/
│   ├── notebooks/
│   ├── jobs/
│   └── clusters.yaml
├── tests/
│   ├── unit_tests.py
│   └── integration_tests.py
└── .github/
    └── workflows/

2. GitHub Actions Workflow for ADF

name: Deploy ADF Pipeline

on:
  push:
    branches: [main]
    paths:
      - 'adf/**'

jobs:
  deploy:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v2
    
    # Build ARM template
    - name: Build ADF ARM Template
      run: |
        pip install azure-cli
        az datafactory build --resource-group ${{ secrets.AZURE_RG }} \
                           --factory-name ${{ secrets.ADF_NAME }} \
                           --output-folder ./adf_arm
    
    # Validate template
    - name: Validate ARM Template
      run: |
        az deployment group validate \
          --resource-group ${{ secrets.AZURE_RG }} \
          --template-file ./adf_arm/ARMTemplateForFactory.json
    
    # Deploy to staging
    - name: Deploy to Staging
      run: |
        az deployment group create \
          --resource-group ${{ secrets.AZURE_RG }} \
          --template-file ./adf_arm/ARMTemplateForFactory.json \
          --parameters ./adf_arm/ARMTemplateParametersForFactory.json
    
    # Run integration tests
    - name: Run Integration Tests
      run: |
        pip install pytest azure-storage-blob
        pytest tests/integration_tests.py

3. Databricks Deployment Using GitHub Actions

name: Deploy Databricks Notebooks

on:
  push:
    branches: [main]
    paths:
      - 'databricks/**'

jobs:
  deploy:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v2
    
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: 3.9
    
    - name: Install Databricks CLI
      run: pip install databricks-cli
    
    - name: Configure Databricks
      run: |
        echo "${{ secrets.DATABRICKS_CONFIG }}" > ~/.databrickscfg
        chmod 600 ~/.databrickscfg
    
    - name: Deploy Notebooks
      run: |
        databricks workspace import-dir ./databricks/notebooks \
          /Shared/Deployments \
          --overwrite
    
    - name: Update Job Configurations
      run: |
        databricks jobs reset --job-id ${{ secrets.DATABRICKS_JOB_ID }} \
          --json-file ./databricks/jobs/job_config.json
    
    - name: Run Notebook Tests
      run: |
        databricks runs submit --new-cluster \
          --spark-python-task notebook-path /Shared/Tests/test_suite

4. Python Unit Test Example

# tests/unit_tests.py
import unittest
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

class TestDataTransformation(unittest.TestCase):
    
    @classmethod
    def setUpClass(cls):
        cls.spark = SparkSession.builder \
            .appName("test") \
            .config("spark.sql.shuffle.partitions", "1") \
            .getOrCreate()
    
    def test_customer_data_processing(self):
        # Arrange
        input_data = [
            (1, "John", "john@example.com"),
            (2, "Jane", "jane@example.com")
        ]
        schema = StructType([
            StructField("id", IntegerType()),
            StructField("name", StringType()),
            StructField("email", StringType())
        ])
        df = self.spark.createDataFrame(input_data, schema)
        
        # Act
        result = df.filter(df.id > 0)
        
        # Assert
        self.assertEqual(result.count(), 2)
        self.assertIn("email", result.columns)

if __name__ == '__main__':
    unittest.main()

5. Terraform for Infrastructure as Code

# infrastructure/main.tf
terraform {
  required_providers {
    azurerm = {
      source  = "hashicorp/azurerm"
      version = "~> 3.0"
    }
  }
}

provider "azurerm" {
  features {}
}

resource "azurerm_data_factory" "adf" {
  name                = var.adf_name
  location            = var.location
  resource_group_name = azurerm_resource_group.rg.name
}

resource "azurerm_databricks_workspace" "databricks" {
  name                = var.databricks_name
  location            = var.location
  resource_group_name = azurerm_resource_group.rg.name
  sku                 = "standard"
}

# Deploy with: terraform apply -var-file="environments/prod.tfvars"

Complete CI/CD Pipeline Flow:

Code Push → GitHub Actions Trigger
  ↓
Source Code Checkout
  ↓
Build/Validate Templates
  ↓
Run Unit Tests
  ↓
Deploy to Dev
  ↓
Run Integration Tests
  ↓
Deploy to Staging
  ↓
Smoke Tests
  ↓
Deploy to Production

---

12. How do you Manage and Monitor ADF Pipeline Performance?

Answer:

Monitoring ADF pipelines involves multiple layers of observation and alerting.

1. Enable Diagnostic Logging

{
  "name": "ADF Diagnostics",
  "type": "Microsoft.Insights/diagnosticSettings",
  "properties": {
    "logs": [
      {
        "category": "PipelineRuns",
        "enabled": true,
        "retentionPolicy": {
          "days": 90,
          "enabled": true
        }
      },
      {
        "category": "ActivityRuns",
        "enabled": true
      }
    ],
    "metrics": [
      {
        "category": "AllMetrics",
        "enabled": true
      }
    ],
    "destination": {
      "workspaceId": "/subscriptions/{subId}/resourcegroups/{rg}/providers/microsoft.operationalinsights/workspaces/{workspace}"
    }
  }
}

2. Monitor Pipeline Runs Using Python

from azure.identity import DefaultAzureCredential
from azure.mgmt.datafactory import DataFactoryManagementClient

# Initialize client
credential = DefaultAzureCredential()
adf_client = DataFactoryManagementClient(credential, subscription_id)

# Get pipeline runs
pipeline_runs = adf_client.pipeline_runs.query_by_factory(
    resource_group_name,
    factory_name,
    {
        "lastUpdatedAfter": datetime(2024, 1, 1),
        "lastUpdatedBefore": datetime.now(),
        "filters": [
            {
                "operand": "PipelineName",
                "operator": "Equals",
                "values": ["MyPipeline"]
            }
        ]
    }
)

for run in pipeline_runs.value:
    print(f"Pipeline: {run.pipeline_name}")
    print(f"Status: {run.status}")
    print(f"Run ID: {run.run_id}")
    print(f"Duration: {(run.run_end - run.run_start).total_seconds()} seconds")

3. Create Alerts Using Application Insights

{
  "name": "ADF Pipeline Failure Alert",
  "type": "Microsoft.Insights/metricAlerts",
  "properties": {
    "description": "Alert when ADF pipeline fails",
    "severity": 2,
    "enabled": true,
    "scopes": ["/subscriptions/{subId}/resourceGroups/{rg}/providers/Microsoft.DataFactory/factories/{adf}"],
    "evaluationFrequency": "PT5M",
    "windowSize": "PT15M",
    "criteria": {
      "odata.type": "Microsoft.Azure.Monitor.MultipleResourceMultipleMetricCriteria",
      "allOf": [
        {
          "name": "Pipeline Failed",
          "metricName": "PipelineFailedRuns",
          "operator": "GreaterThan",
          "threshold": 0,
          "timeAggregation": "Total"
        }
      ]
    },
    "actions": [
      {
        "actionGroupId": "/subscriptions/{subId}/resourceGroups/{rg}/providers/Microsoft.Insights/actionGroups/{actionGroup}"
      }
    ]
  }
}

4. KQL Queries for Log Analytics

// Query 1: Failed activities in last 24 hours
ActivityRuns
| where TimeGenerated > ago(24h)
| where Status == "Failed"
| summarize Count=count() by ActivityName, Status
| order by Count desc

// Query 2: Pipeline execution time trends
PipelineRuns
| where TimeGenerated > ago(7d)
| extend Duration = RunEnd - RunStart
| summarize AvgDuration = avg(Duration), MaxDuration = max(Duration) 
  by PipelineName, bin(TimeGenerated, 1d)

// Query 3: Activity failure rate
ActivityRuns
| where TimeGenerated > ago(7d)
| summarize Total = count(), Failed = countif(Status == "Failed") 
  by ActivityName
| extend FailureRate = (Failed * 100.0) / Total
| where FailureRate > 5

5. Custom Monitoring Dashboard

{
  "name": "ADF Performance Dashboard",
  "type": "Microsoft.Portal/dashboards",
  "properties": {
    "lenses": {
      "0": {
        "order": 0,
        "parts": {
          "0": {
            "position": {
              "x": 0,
              "y": 0,
              "colSpan": 4,
              "rowSpan": 4
            },
            "metadata": {
              "inputs": [
                {
                  "name": "sharedTimeRange",
                  "isOptional": true
                }
              ],
              "type": "Extension/Microsoft_Azure_Monitoring/PartType/MetricsChartPart",
              "settings": {
                "content": {
                  "metrics": [
                    {
                      "resourceMetadata": {
                        "id": "/subscriptions/{subId}/resourcegroups/{rg}/providers/Microsoft.DataFactory/factories/{adf}"
                      },
                      "name": "PipelineSucceededRuns"
                    }
                  ]
                }
              }
            }
          }
        }
      }
    }
  }
}

6. Programmatic Alert Handling

import smtplib
from email.mime.text import MIMEText

def send_alert_on_failure(pipeline_name, error_message):
    # Configure SMTP
    smtp_server = "smtp.gmail.com"
    sender_email = "alerts@company.com"
    receiver_email = "team@company.com"
    
    # Create message
    message = MIMEText(f"Pipeline {pipeline_name} failed: {error_message}")
    message["Subject"] = f"ADF Alert: {pipeline_name} Failed"
    message["From"] = sender_email
    message["To"] = receiver_email
    
    # Send
    with smtplib.SMTP_SSL(smtp_server, 465) as server:
        server.login(sender_email, "password")
        server.sendmail(sender_email, receiver_email, message.as_string())

---

15. How do you Implement Schema Drift Handling in ADF?

Answer:

Schema drift refers to unexpected changes in source data structure that aren’t captured in the dataset schema definition.

Method 1: Enable Schema Drift in Copy Activity

{
    "name": "CopyActivityWithSchemaDrift",
    "type": "Copy",
    "inputs": [
        {
            "referenceName": "SourceDataset",
            "type": "DatasetReference"
        }
    ],
    "outputs": [
        {
            "referenceName": "SinkDataset",
            "type": "DatasetReference"
        }
    ],
    "typeProperties": {
        "source": {
            "type": "JsonSource"
        },
        "sink": {
            "type": "ParquetSink"
        },
        "enableSchemaDrift": true,
        "translator": {
            "type": "TabularTranslator",
            "mappings": []  // Empty mappings allow auto-detection
        }
    }
}

Method 2: Use Data Flows for Schema Flexibility

{
    "name": "SchemaFlexibleDataFlow",
    "type": "ExecuteDataFlow",
    "typeProperties": {
        "dataflow": {
            "referenceName": "FlexibleTransformationFlow",
            "type": "DataFlowReference"
        }
    }
}

Data Flow Script:

source(output(
          col1 as string,
          col2 as integer
     ),
     allowSchemaDrift: true,
     validateSchema: false) ~> Source

Source select(each(match(type!='unknown'), $$ as $$),
              each(match(type=='unknown'), 'unmapped_' + $$ as $$)) ~> CaptureNewColumns

CaptureNewColumns sink() ~> Sink

Method 3: Capture Schema Changes in Metadata Table

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType
import json
from datetime import datetime

spark = SparkSession.builder.appName("SchemaDriftTracking").getOrCreate()

def track_schema_changes(df, source_name):
    # Current schema
    current_schema = df.schema.jsonValue()
    
    # Read previous schema if exists
    try:
        previous_df = spark.read.json(f"/mnt/metadata/{source_name}_schema.json")
        previous_schema = previous_df.select("schema").collect()[0][0]
    except:
        previous_schema = None
    
    # Compare and log changes
    if previous_schema != current_schema:
        change_log = spark.createDataFrame([
            {
                "source": source_name,
                "timestamp": datetime.now(),
                "previous_schema": json.dumps(previous_schema),
                "current_schema": json.dumps(current_schema),
                "new_columns": list(set(current_schema.keys()) - set(previous_schema.keys() if previous_schema else []))
            }
        ])
        
        change_log.write.mode("append").json(f"/mnt/metadata/{source_name}_schema_changes")
    
    # Save current schema
    spark.createDataFrame([{"schema": current_schema}]).write.mode("overwrite").json(f"/mnt/metadata/{source_name}_schema.json")

# Usage
source_df = spark.read.csv("/mnt/raw/data.csv", header=True)
track_schema_changes(source_df, "customer_data")

Method 4: Mapping with Unknown Column Handling

{
    "type": "Copy",
    "typeProperties": {
        "source": {
            "type": "DelimitedTextSource"
        },
        "sink": {
            "type": "JsonSink"
        },
        "translator": {
            "type": "TabularTranslator",
            "columnMappings": [
                {
                    "source": "customer_id",
                    "sink": "customer_id"
                }
            ],
            "schemaMapping": {
                "unknownColumn": "$",  // Map unknown columns to root
                "allowSchemaEvolution": true
            }
        }
    }
}

---

24. Explain the Use of Integration Runtime (IR) in ADF

Answer:

Integration Runtime (IR) is the compute infrastructure that powers ADF activities. It’s the bridge between ADF and data sources.

Types of Integration Runtime:

1. Azure Integration Runtime (Cloud)

Used for connecting to cloud services and public endpoints:

{
    "name": "AzureIR",
    "type": "Microsoft.DataFactory/factories/integrationruntimes",
    "properties": {
        "type": "Managed",
        "typeProperties": {
            "computeProperties": {
                "location": "AutoResolve"
            }
        }
    }
}

Best for:

• Cloud data sources (Azure Storage, Cosmos DB)
• SaaS applications
• Public data sources
• Scenarios where data doesn’t cross firewalls

2. Self-Hosted Integration Runtime

For connecting to on-premises and private network resources:

# Install Self-Hosted IR on premises
# Download and run MSI from Azure Portal

# Configure with gateway (after installation)
$gateway = New-AzDataFactoryV2IntegrationRuntime -ResourceGroupName "MyRG" `
    -DataFactoryName "MyADF" `
    -Name "SelfHostedIR" `
    -Type SelfHosted

Best for:

• On-premises SQL Server, Oracle, DB2
• Private networks behind firewalls
• Hybrid scenarios
• Complex data transformations requiring local compute

3. Azure-SSIS Integration Runtime

Dedicated for SSIS package execution:

{
    "name": "SSISIntegrationRuntime",
    "type": "Microsoft.DataFactory/factories/integrationruntimes",
    "properties": {
        "type": "Managed",
        "managedVirtualNetwork": {
            "referenceName": "default",
            "type": "ManagedVirtualNetworkReference"
        },
        "typeProperties": {
            "ssisProperties": {
                "catalogInfo": {
                    "catalogServerEndpoint": "myserver.database.windows.net",
                    "catalogAdminUserName": "sqladmin",
                    "catalogAdminPassword": {
                        "type": "SecureString",
                        "value": "password"
                    },
                    "catalogPricingTier": "Standard"
                },
                "licenseType": "LicenseIncluded"
            }
        }
    }
}

Best for:

• Running SSIS packages
• Complex ETL processes
• Backward compatibility with SSIS

IR Comparison Table:

Feature	Azure IR	Self-Hosted IR	SSIS IR
Cloud Services	✓	✗	Limited
On-Premises	✗	✓	✗
SSIS Packages	✗	Limited	✓
Cost	Per activity	Fixed + hardware	Fixed
Latency	Low (cloud)	Variable	Low
Setup	Instant	Manual	Complex

4. Practical Example: Using Different IRs

{
    "name": "HybridDataPipeline",
    "properties": {
        "activities": [
            {
                "name": "CopyFromAzureBlob",
                "type": "Copy",
                "inputs": [
                    {
                        "referenceName": "AzureBlobDataset",
                        "type": "DatasetReference"
                    }
                ],
                "outputs": [
                    {
                        "referenceName": "StagingDataset",
                        "type": "DatasetReference"
                    }
                ],
                "linkedServiceName": {
                    "referenceName": "AzureStorageLinkedService",
                    "type": "LinkedServiceReference"
                },
                "typeProperties": {
                    "integrationRuntime": {
                        "referenceName": "AzureIR",
                        "type": "IntegrationRuntimeReference"
                    }
                }
            },
            {
                "name": "CopyToOnPremiseSQL",
                "type": "Copy",
                "dependsOn": [
                    {
                        "activity": "CopyFromAzureBlob",
                        "dependencyConditions": ["Succeeded"]
                    }
                ],
                "inputs": [
                    {
                        "referenceName": "StagingDataset",
                        "type": "DatasetReference"
                    }
                ],
                "outputs": [
                    {
                        "referenceName": "OnPremisesSQLDataset",
                        "type": "DatasetReference"
                    }
                ],
                "linkedServiceName": {
                    "referenceName": "OnPremisesSQLLinkedService",
                    "type": "LinkedServiceReference"
                },
                "typeProperties": {
                    "integrationRuntime": {
                        "referenceName": "SelfHostedIR",
                        "type": "IntegrationRuntimeReference"
                    }
                }
            }
        ]
    }
}

5. Monitoring Integration Runtime

from azure.identity import DefaultAzureCredential
from azure.mgmt.datafactory import DataFactoryManagementClient

credential = DefaultAzureCredential()
adf_client = DataFactoryManagementClient(credential, subscription_id)

# Get IR status
ir = adf_client.integration_runtimes.get(
    resource_group_name,
    factory_name,
    "SelfHostedIR"
)

print(f"IR State: {ir.state}")
print(f"IR Type: {ir.type}")

# Monitor node status (for self-hosted)
nodes = adf_client.integration_runtimes.list_auth_keys(
    resource_group_name,
    factory_name,
    "SelfHostedIR"
)

---

Data Engineering Patterns

14. Explain the Concept of Delta Lake and its Advantages

Answer:

Delta Lake is an open-source storage layer that brings ACID (Atomicity, Consistency, Isolation, Durability) transactions to Apache Spark and big data workloads.

Core Concept:

Delta Lake adds a transaction log to Parquet files, enabling:

• ACID transactions on data lake
• Schema enforcement
• Time travel
• Unified batch and streaming

Traditional Parquet:
File1.parquet → File2.parquet → File3.parquet (No transaction guarantees)

Delta Lake:
┌─────────────────────────────────┐
│   Delta Table Files             │
├─────────────────────────────────┤
│ ├── part-00001.parquet          │
│ ├── part-00002.parquet          │
│ └── _delta_log/                 │
│     ├── 00000000000000.json     │ ← Transaction Log
│     ├── 00000000000001.json     │
│     └── 00000000000002.json     │
└─────────────────────────────────┘

Key Advantages:

1. ACID Transactions

from delta.tables import DeltaTable

# Multiple transactions with guaranteed consistency
df1 = spark.createDataFrame([(1, "Alice", 50000)])
df1.write.format("delta").mode("overwrite").save("/mnt/delta/employees")

df2 = spark.createDataFrame([(2, "Bob", 60000)])
df2.write.format("delta").mode("append").save("/mnt/delta/employees")

# Concurrent writes are safely handled
# All or nothing - no partial updates

2. Schema Enforcement

# Write with schema
schema = "id INT, name STRING, salary INT"
df.write.format("delta").mode("overwrite").option("mergeSchema", "false").save(path)

# Attempting to write incompatible data fails
bad_df = spark.createDataFrame([(1, "Alice", "not_a_number")])
bad_df.write.format("delta").mode("append").save(path)  # Raises error

3. Time Travel (Temporal Queries)

# Read data from specific time point
df_past = spark.read.format("delta").option("versionAsOf", 0).load(path)

# Read from timestamp
df_historical = spark.read \
    .format("delta") \
    .option("timestampAsOf", "2024-01-15 10:30:00") \
    .load(path)

# List all versions
delta_table = DeltaTable.forPath(spark, path)
delta_table.history().show()

4. Unified Batch and Streaming

# Streaming write to Delta
streaming_df = spark.readStream.csv("/mnt/streaming_source")
streaming_df.writeStream \
    .format("delta") \
    .mode("append") \
    .option("checkpointLocation", "/mnt/checkpoint") \
    .start("/mnt/delta/streaming_table")

# Batch query the same table
batch_df = spark.read.format("delta").load("/mnt/delta/streaming_table")
batch_df.show()

5. Data Optimization

delta_table = DeltaTable.forPath(spark, "/mnt/delta/data")

# Compact small files
delta_table.optimize().executeCompaction()

# Z-order for optimized queries
delta_table.optimize().executeZorderBy("customer_id")

# Vacuum to remove old versions
delta_table.vacuum(168)  # Keep 7 days of versions

6. Merge for Upserts

# Efficient upsert operation
delta_table = DeltaTable.forPath(spark, path)

delta_table.merge(
    source_df, 
    "target.id = source.id"
).whenMatchedUpdate(set={"salary": col("source.salary")}) \
 .whenNotMatchedInsert(values={"id": col("source.id"), "name": col("source.name"), "salary": col("source.salary")}) \
 .execute()

Comparison with Iceberg and Hudi:

Feature	Delta Lake	Apache Iceberg	Apache Hudi
ACID Support	✓	✓	✓
Time Travel	✓	✓	✓
Streaming	✓	Limited	✓
Performance	Excellent	Excellent	Good
Ecosystem	Databricks	Netflix	Uber

---

17. What is the Significance of Z-Ordering in Delta Tables?

Answer:

Z-Ordering is a data optimization technique in Delta Lake that physically reorders data to improve query performance on multiple columns.

How Z-Ordering Works:

Z-ordering uses a space-filling curve algorithm to map multidimensional data into one dimension while preserving spatial locality. This clusters related data together.

from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/mnt/delta/sales_data")

# Z-order by multiple columns
delta_table.optimize().executeZorderBy("region", "product_id", "date")

Before Z-Ordering:

File 1: [Region=US, Dates=2024-01-15 to 2024-02-20]
File 2: [Region=EU, Dates=2024-01-01 to 2024-03-15]
File 3: [Region=US, Dates=2024-02-25 to 2024-04-10]

Query: WHERE region='US' AND date='2024-02-10'
→ Must scan Files 1 and 3 (full files)

After Z-Ordering:

File 1: [Region=US, Dates=2024-01-15 to 2024-02-20, Product=Electronics]
File 2: [Region=US, Dates=2024-01-15 to 2024-02-20, Product=Furniture]
File 3: [Region=US, Dates=2024-02-25 to 2024-04-10, Product=Electronics]

Query: WHERE region='US' AND date='2024-02-10'
→ Can skip File 3, scans only Files 1-2

Practical Benefits:

1. Reduced I/O – Skips unnecessary files during queries
2. Better Parallelism – Partition pruning works more effectively
3. Lower Query Latency – Especially for multi-column filters
4. Data Locality – Related data stored together

Complete Example:

from pyspark.sql import SparkSession
from delta.tables import DeltaTable

spark = SparkSession.builder.appName("ZOrderExample").getOrCreate()

# Create sample data
sales_data = spark.createDataFrame([
    (1, "US", "Laptop", 1200, "2024-01-15"),
    (2, "EU", "Mouse", 25, "2024-01-20"),
    (3, "US", "Keyboard", 85, "2024-02-10"),
    (4, "APAC", "Monitor", 350, "2024-02-15"),
    (5, "US", "Laptop", 1200, "2024-02-25"),
], ["id", "region", "product", "price", "date"])

# Write as Delta table
sales_data.write.format("delta").mode("overwrite").save("/mnt/delta/sales")

# Apply Z-Ordering
delta_table = DeltaTable.forPath(spark, "/mnt/delta/sales")
delta_table.optimize().executeZorderBy("region", "product")

# Now queries benefit from Z-ordering
result = spark.read.format("delta").load("/mnt/delta/sales") \
    .filter("region='US' AND product='Laptop'") \
    .show()

# Check optimization history
delta_table.history().select("version", "operation", "operationMetrics").show()

When to Use Z-Ordering:

• ✓ Columns frequently used together in WHERE clauses
• ✓ Large tables (100GB+) where I/O optimization matters
• ✓ OLAP queries with multiple filtering dimensions
• ✗ Small tables or frequently updated data
• ✗ Columns with extreme skewness

Performance Impact:

Typical improvements:

• Query latency: 20-70% reduction
• Bytes scanned: 30-50% reduction
• At cost of: 5-15% write overhead

---

18. How do you Handle Incremental Data Load in Databricks?

Answer:

Incremental loading only processes new or changed data rather than reprocessing the entire dataset.

Method 1: Using Modified Date Watermarking

from pyspark.sql import SparkSession
from pyspark.sql.functions import max as spark_max
import json

spark = SparkSession.builder.appName("IncrementalLoad").getOrCreate()

# Track watermark (last processed timestamp)
watermark_file = "/mnt/metadata/load_watermark.json"

def read_watermark():
    try:
        with open(watermark_file) as f:
            return json.load(f)["last_processed"]
    except:
        return "1900-01-01"

def write_watermark(timestamp):
    with open(watermark_file, 'w') as f:
        json.dump({"last_processed": str(timestamp)}, f)

# Read data newer than watermark
last_processed = read_watermark()
incremental_df = spark.read.csv("/mnt/raw/data.csv") \
    .filter(f"modified_date > '{last_processed}'")

# Process data
processed_df = incremental_df.transform(your_transformation_function)

# Write to Delta with merge
delta_table = DeltaTable.forPath(spark, "/mnt/delta/processed_data")

delta_table.merge(
    processed_df,
    "target.id = source.id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# Update watermark
new_watermark = processed_df.agg(spark_max("modified_date")).collect()[0][0]
write_watermark(new_watermark)

Method 2: CDC (Change Data Capture)

# For SQL Server with CDC enabled
cdc_df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:sqlserver://server:1433;database=db") \
    .option("dbtable", "cdc.fn_cdc_get_all_changes_table_name") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

# Process CDC records
from pyspark.sql.functions import col

inserts = cdc_df.filter(col("__$operation") == 2)
updates = cdc_df.filter(col("__$operation") == 4)
deletes = cdc_df.filter(col("__$operation") == 1)

# Apply changes
delta_table.merge(
    inserts.select("id", "name", "value"),
    "target.id = source.id"
).whenNotMatchedInsertAll() \
 .execute()

delta_table.merge(
    updates.select("id", "name", "value"),
    "target.id = source.id"
).whenMatchedUpdateAll() \
 .execute()

deletes_df = deletes.select("id")
delta_table.delete(f"id IN (SELECT id FROM deletes_df)")

Method 3: Partition-Based Incremental Load

from pyspark.sql.functions import year, month, to_date
from datetime import datetime, timedelta

# Load only latest partitions
last_date = spark.sql("SELECT MAX(date) FROM delta_table").collect()[0][0]
cutoff_date = last_date - timedelta(days=1)

raw_data = spark.read.csv("/mnt/raw") \
    .filter(f"date >= '{cutoff_date}'")

# Repartition by date for efficient storage
processed_df = raw_data \
    .transform(transformation) \
    .withColumn("year", year("date")) \
    .withColumn("month", month("date"))

# Write with partitioning
processed_df.write \
    .format("delta") \
    .partitionBy("year", "month") \
    .mode("overwrite") \
    .save("/mnt/delta/partitioned_data")

Method 4: Spark Structured Streaming for Real-Time Incremental

from pyspark.sql.streaming import StreamingQuery

# Read from streaming source
streaming_df = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "csv") \
    .option("cloudFiles.schemaLocation", "/mnt/schema") \
    .load("/mnt/incoming/")

# Transform
transformed_df = streaming_df \
    .filter("status = 'valid'") \
    .select("id", "name", "value")

# Write incrementally to Delta
query = transformed_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/checkpoint") \
    .start("/mnt/delta/streaming_data")

# Monitor
query.awaitTermination()

Method 5: Notebook with Incremental Framework

# Complete incremental load notebook
%run ./utilities/common_functions

# Configuration
SOURCE_PATH = "/mnt/raw/customer_data"
TARGET_PATH = "/mnt/delta/customers"
WATERMARK_PATH = "/mnt/metadata/customer_watermark.json"

# Step 1: Read watermark
last_processed = read_watermark(WATERMARK_PATH)
print(f"Loading data modified after: {last_processed}")

# Step 2: Read incremental data
incremental_df = spark.read \
    .format("csv") \
    .option("header", "true") \
    .load(SOURCE_PATH) \
    .filter(f"updated_at > '{last_processed}'")

# Step 3: Validate
print(f"Records to process: {incremental_df.count()}")

# Step 4: Transform
cleaned_df = incremental_df \
    .dropDuplicates(["id"]) \
    .na.drop()

# Step 5: Merge
merge_incremental(spark, cleaned_df, TARGET_PATH, "id")

# Step 6: Update watermark
update_watermark(WATERMARK_PATH, cleaned_df, "updated_at")

print("Incremental load completed successfully")

---

20. How do you Optimize Data Partitioning in ADLS?

Answer:

Optimal partitioning strategy significantly improves query performance and cost efficiency.

Partitioning Strategy Selection:

Decision Tree:
├── Small files problem?
│   └── YES → Date-based partitioning
├── Query patterns known?
│   └── YES → Partition by frequently filtered columns
├── Data skew expected?
│   └── YES → Partition by range instead of hash
└── Update frequency high?
    └── YES → Partition by date for easier retention

1. Date-Based Partitioning (Most Common)

from pyspark.sql.functions import year, month, day

data = spark.read.format("csv").option("header", "true").load("/mnt/raw/")

# Add date columns
partitioned_data = data \
    .withColumn("year", year("transaction_date")) \
    .withColumn("month", month("transaction_date")) \
    .withColumn("day", day("transaction_date"))

# Write with partitioning
partitioned_data.write \
    .format("parquet") \
    .partitionBy("year", "month", "day") \
    .mode("overwrite") \
    .save("/mnt/processed/transactions")

# Directory structure created:
# /mnt/processed/transactions/year=2024/month=1/day=15/part-00000.parquet
# /mnt/processed/transactions/year=2024/month=1/day=16/part-00000.parquet

2. Multi-Column Partitioning for Dimensionality

# Partition by region and product type
data.write \
    .format("delta") \
    .partitionBy("region", "product_category") \
    .mode("overwrite") \
    .save("/mnt/delta/sales")

# This enables efficient queries
query = spark.read.format("delta").load("/mnt/delta/sales") \
    .filter("region = 'US' AND product_category = 'Electronics'")
# Only reads relevant partitions

3. Bucketing for Further Optimization

# Partition then bucket for even better performance
data.write \
    .format("delta") \
    .partitionBy("date") \
    .bucketBy(10, "customer_id") \
    .mode("overwrite") \
    .save("/mnt/delta/bucketted_data")

# Benefits join performance:
# df1.join(df2, "customer_id") runs faster

4. Range-Based Partitioning for Skewed Data

from pyspark.sql.window import Window
from pyspark.sql.functions import ntile

# Identify skew
data.groupBy("category").count().orderBy("count", ascending=False).show()

# Create range buckets instead of raw values
data_with_bucket = data.withColumn(
    "category_bucket",
    ntile(5).over(Window.partitionBy("category").orderBy("value"))
)

data_with_bucket.write \
    .format("delta") \
    .partitionBy("date", "category_bucket") \
    .mode("overwrite") \
    .save("/mnt/delta/balanced_data")

5. Right-Sizing Partitions

# Avoid too many small files
# Rule of thumb: 128MB - 1GB per partition

# Too many partitions (Bad)
data.write.partitionBy("id").save(path)  # Millions of tiny partitions

# Optimized
data.coalesce(num_partitions).write \
    .partitionBy("region", "year", "month") \
    .save(path)

6. Compacting and Reorganizing

from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/mnt/delta/data")

# Compact small files
delta_table.optimize().executeCompaction()

# Z-order for multiple column queries
delta_table.optimize().executeZorderBy("customer_id", "transaction_date")

# Vacuum old versions (retention policy)
delta_table.vacuum(7*24)  # Keep 7 days of data

7. Partition Pruning Example

# Good query - leverages partitioning
result = spark.read.format("delta").load("/mnt/delta/sales") \
    .filter("year = 2024 AND month = 1") \
    .collect()
# Spark only reads year=2024/month=1/ partitions

# Bad query - no partition pruning
result = spark.read.format("delta").load("/mnt/delta/sales") \
    .collect() \
    .filter("year == 2024")
# Reads entire dataset first!

# Check actual partitions scanned
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.sql("""
    SELECT * FROM delta.`/mnt/delta/sales`
    WHERE year = 2024 AND month = 1
""").explain()

8. Monitoring Partition Performance

# Check partition distribution
spark.read.format("delta").load("/mnt/delta/sales") \
    .groupBy("year", "month") \
    .count() \
    .orderBy("count", ascending=False) \
    .show()

# Identify skewed partitions
partition_stats = spark.sql("""
    SELECT year, month, COUNT(*) as record_count
    FROM delta.`/mnt/delta/sales`
    GROUP BY year, month
    HAVING record_count < 1000  -- Too small
    OR record_count > 10000000  -- Too large
""")
partition_stats.show()

---

Additional Utilities

16. Write Python Code to Check if a Number is a Palindrome

Answer:

# Method 1: String Conversion (Simple)
def is_palindrome_string(n):
    """Check if number is palindrome using string"""
    s = str(abs(n))  # abs() to handle negative numbers
    return s == s[::-1]

# Test cases
print(is_palindrome_string(121))    # True
print(is_palindrome_string(123))    # False
print(is_palindrome_string(-121))   # True
print(is_palindrome_string(0))      # True


# Method 2: Mathematical Approach (No String Conversion)
def is_palindrome_math(n):
    """Check if palindrome using math operations"""
    if n < 0:
        n = abs(n)
    
    original = n
    reversed_num = 0
    
    while n > 0:
        digit = n % 10
        reversed_num = reversed_num * 10 + digit
        n = n // 10
    
    return original == reversed_num

# Test cases
print(is_palindrome_math(121))    # True
print(is_palindrome_math(1001))   # True
print(is_palindrome_math(1234))   # False


# Method 3: Two-Pointer Approach with String
def is_palindrome_two_pointer(n):
    """Check using two pointers"""
    s = str(abs(n))
    left, right = 0, len(s) - 1
    
    while left < right:
        if s[left] != s[right]:
            return False
        left += 1
        right -= 1
    
    return True

# Test cases
print(is_palindrome_two_pointer(9009))  # True
print(is_palindrome_two_pointer(1000))  # False


# Method 4: One-Liner with List Comprehension
def is_palindrome_oneliner(n):
    """Concise palindrome check"""
    s = str(abs(n))
    return all(s[i] == s[-(i+1)] for i in range(len(s)//2))

# Test cases
print(is_palindrome_oneliner(12321))  # True


# Method 5: Handle Multiple Numbers (List)
def check_palindromes(numbers):
    """Check multiple numbers"""
    results = {}
    for num in numbers:
        s = str(abs(num))
        results[num] = s == s[::-1]
    return results

# Test
numbers = [121, 123, 1001, -99, 0]
print(check_palindromes(numbers))
# Output: {121: True, 123: False, 1001: True, -99: True, 0: True}


# Performance Comparison
if __name__ == "__main__":
    import timeit
    
    test_num = 123454321
    
    # Time each method
    time_string = timeit.timeit(
        lambda: is_palindrome_string(test_num),
        number=100000
    )
    print(f"String method: {time_string:.4f}s")
    
    time_math = timeit.timeit(
        lambda: is_palindrome_math(test_num),
        number=100000
    )
    print(f"Math method: {time_math:.4f}s")
    
    time_two_ptr = timeit.timeit(
        lambda: is_palindrome_two_pointer(test_num),
        number=100000
    )
    print(f"Two-pointer method: {time_two_ptr:.4f}s")

Complexity Analysis:

Method	Time	Space	Notes
String	O(n)	O(n)	Simple, creates new string
Math	O(log n)	O(1)	Efficient, no extra space
Two-Pointer	O(n)	O(n)	Uses string, clear logic
Oneliner	O(n)	O(n)	Pythonic but less readable

---

21. Describe the Process of Creating a Data Pipeline for Real-Time Analytics

Answer:

A real-time analytics pipeline processes streaming data and delivers insights within seconds.

Architecture:

Data Source → Message Queue → Stream Processor → Storage → Real-Time Dashboard
    ↓              ↓                  ↓             ↓              ↓
  IoT/Logs     Kafka/Event Hub   Spark/Flink   Delta Lake    Power BI/Grafana

Step 1: Data Ingestion

# Option A: Azure Event Hubs
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("RealTimeAnalytics").getOrCreate()

connection_string = "Endpoint=sb://namespace.servicebus.windows.net/;..."
consumer_group = "$Default"

stream_df = spark \
    .readStream \
    .format("eventhubs") \
    .options(**{
        "eventhubs.connectionString": connection_string,
        "eventhubs.consumerGroup": consumer_group,
        "maxEventsPerTrigger": 5000,
        "eventhubs.startingPosition": """{"offset": "-1", "seqNo": -1, "enqueuedTime": null}"""
    }) \
    .load()

Step 2: Parse and Transform

from pyspark.sql.functions import from_json, col, window, avg, count
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Define schema for incoming messages
schema = StructType([
    StructField("sensor_id", StringType()),
    StructField("temperature", IntegerType()),
    StructField("humidity", IntegerType()),
    StructField("timestamp", StringType())
])

# Parse JSON
parsed_df = stream_df \
    .select(from_json(col("body").cast(StringType()), schema).alias("data")) \
    .select("data.*") \
    .withColumn("timestamp", col("timestamp").cast("timestamp"))

# Apply transformations
transformed_df = parsed_df \
    .filter(col("temperature") > -50) \
    .filter(col("temperature") < 150)

Step 3: Aggregation and Feature Engineering

# Real-time aggregations
aggregated_df = transformed_df \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window(col("timestamp"), "5 minutes", "1 minute"),  # 5-min windows, 1-min slide
        col("sensor_id")
    ) \
    .agg(
        avg("temperature").alias("avg_temp"),
        count("*").alias("message_count")
    ) \
    .select(
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        "sensor_id",
        "avg_temp",
        "message_count"
    )

Step 4: Anomaly Detection

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import IsolationForest

# Prepare features for anomaly detection
vectorized_df = aggregated_df \
    .select(
        "sensor_id",
        "timestamp",
        "avg_temp",
        VectorAssembler(
            inputCols=["avg_temp"],
            outputCol="features"
        ).transform(aggregated_df)
    )

# Simple threshold-based anomaly detection
anomalies = transformed_df \
    .withColumn("is_anomaly", 
        (col("temperature") > 40) | (col("humidity") > 85)
    ) \
    .filter(col("is_anomaly") == True)

Step 5: Output to Multiple Sinks

# Sink 1: Real-time Dashboard (in-memory)
query_memory = aggregated_df \
    .writeStream \
    .format("memory") \
    .outputMode("update") \
    .queryName("sensor_metrics") \
    .option("checkpointLocation", "/tmp/checkpoint_memory") \
    .start()

# Register for SQL queries
spark.sql("""
    SELECT * FROM sensor_metrics 
    WHERE avg_temp > 35
    ORDER BY window_start DESC
""").show()

# Sink 2: Delta Lake (persistent storage)
query_delta = aggregated_df \
    .writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/checkpoint/sensor_metrics") \
    .option("mergeSchema", "true") \
    .start("/mnt/delta/sensor_analytics")

# Sink 3: Alert System
query_alerts = anomalies \
    .select("sensor_id", "temperature", "timestamp") \
    .writeStream \
    .foreachBatch(lambda batch_df, batch_id: send_alerts(batch_df)) \
    .option("checkpointLocation", "/mnt/checkpoint/alerts") \
    .start()

# Custom function for alerts
def send_alerts(batch_df):
    for row in batch_df.collect():
        if row.temperature > 50:
            # Send alert
            send_email(f"High temperature alert: {row.temperature}°C at {row.sensor_id}")

Step 6: Monitoring

# Monitor active streams
for stream in spark.streams.active:
    print(f"Stream: {stream.name}")
    print(f"Status: {stream.status}")
    print(f"Recent Progress: {stream.lastProgress}")

# Define SLA checks
def check_latency(batch_df):
    import time
    current_time = time.time()
    max_timestamp = batch_df.select(max("timestamp")).collect()[0][0]
    latency = (current_time - max_timestamp.timestamp()) / 1000
    
    if latency > 5:  # 5 seconds SLA
        alert(f"Pipeline latency exceeded: {latency}ms")

query.awaitTermination()

Complete Real-Time Notebook:

# %run ./utilities/streaming_functions

# Configuration
CHECKPOINT_PATH = "/mnt/checkpoints/"
ALERT_THRESHOLD_TEMP = 45
ALERT_THRESHOLD_HUMIDITY = 80

# Initialize
spark.conf.set("spark.sql.shuffle.partitions", "4")
spark.conf.set("spark.sql.streaming.schemaInference", "true")

# Read from source
source_stream = spark.readStream \
    .format("eventhubs") \
    .options(**event_hub_options) \
    .load()

# Transform
transformed = parse_and_transform(source_stream)

# Detect anomalies
alerts = transformed \
    .filter(
        (col("temperature") > ALERT_THRESHOLD_TEMP) |
        (col("humidity") > ALERT_THRESHOLD_HUMIDITY)
    )

# Write to storage
alerts.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", f"{CHECKPOINT_PATH}alerts") \
    .start("/mnt/delta/alerts")

# Write aggregates
aggregates = transformed \
    .groupBy(window("timestamp", "1 minute"), "sensor_id") \
    .agg(avg("temperature"), avg("humidity"))

aggregates.writeStream \
    .format("delta") \
    .outputMode("update") \
    .option("checkpointLocation", f"{CHECKPOINT_PATH}metrics") \
    .start("/mnt/delta/metrics")

print("Real-time analytics pipeline started")

---

25. How do you Design a Fault-Tolerant Architecture for Big Data Processing?

Answer:

Fault tolerance ensures that data pipelines continue operating or recover gracefully when failures occur.

1. Checkpoint and Recovery Strategy

# Structured Streaming with checkpoints
query = stream_df \
    .writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/checkpoints/streaming_job") \
    .start("/mnt/delta/data")

# Checkpoint stores:
# - Offset information
# - Batch metadata
# - State information
# - Allows restart from exact point of failure

# Verify checkpoint
import os
checkpoint_path = "/mnt/checkpoints/streaming_job"
offsets = os.listdir(checkpoint_path)
print(f"Checkpoint has {len(offsets)} batches")

2. Idempotent Operations

# Good - Idempotent (safe to retry)
df.write \
    .format("delta") \
    .mode("overwrite") \  # Replaces completely
    .option("replaceWhere", "date = '2024-01-15'") \
    .save(path)

# Bad - Not idempotent
df.write \
    .format("delta") \
    .mode("append") \  # Could duplicate if retried
    .save(path)

# Better - Idempotent merge
delta_table = DeltaTable.forPath(spark, path)
delta_table.merge(
    df,
    "target.id = source.id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()  # Can be safely retried

3. Distributed State Management

# Use Delta Lake for distributed state
state_table = spark.createDataFrame([
    (1, "Processing", "2024-01-15"),
    (2, "Completed", "2024-01-15")
], ["job_id", "status", "date"])

state_table.write \
    .format("delta") \
    .mode("overwrite") \
    .option("location", "/mnt/delta/job_state") \
    .save()

# Check before processing
def safe_process_job(job_id):
    state = spark.sql(f"SELECT status FROM delta_job_state WHERE job_id = {job_id}")
    if state.filter("status = 'Completed'").count() > 0:
        print(f"Job {job_id} already processed")
        return
    
    # Process
    process_job(job_id)
    
    # Update state
    spark.sql(f"UPDATE delta_job_state SET status = 'Completed' WHERE job_id = {job_id}")

4. Retry Logic with Exponential Backoff

import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1):
    """Implement exponential backoff retry"""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s...")
            time.sleep(delay)

# Usage
def read_data():
    return spark.read.csv("/mnt/unreliable_source/data.csv")

data = retry_with_backoff(read_data)

5. Circuit Breaker Pattern

from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = 1      # Normal operation
    OPEN = 2        # Failing, refuse requests
    HALF_OPEN = 3   # Testing recovery

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
    
    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            # Check if we can try recovery
            if datetime.now() - self.last_failure_time > timedelta(seconds=self.recovery_timeout):
                self.state = CircuitState.HALF_OPEN
                print("Circuit breaker: attempting recovery...")
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func(*args, **kwargs)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise

    def on_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        print("Circuit breaker: CLOSED")

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"Circuit breaker: OPEN (failures: {self.failure_count})")

# Usage
breaker = CircuitBreaker(failure_threshold=3)

def unreliable_read():
    return spark.read.csv("/mnt/source/")

try:
    data = breaker.call(unreliable_read)
except Exception as e:
    print(f"Failed to read: {e}")

6. Multi-Region Failover

{
    "name": "MultiRegionPipeline",
    "properties": {
        "activities": [
            {
                "name": "TryPrimaryRegion",
                "type": "Copy",
                "inputs": [{"referenceName": "PrimarySource"}],
                "outputs": [{"referenceName": "Destination"}],
                "onFailure": [
                    {
                        "dependencyConditions": ["Failed"],
                        "activity": "TrySecondaryRegion"
                    }
                ]
            },
            {
                "name": "TrySecondaryRegion",
                "type": "Copy",
                "inputs": [{"referenceName": "SecondarySource"}],
                "outputs": [{"referenceName": "Destination"}]
            }
        ]
    }
}

7. Data Validation and Quality Checks

# Quality check framework
def validate_data_quality(df, checks):
    """
    Run quality checks on data
    checks: list of (check_name, check_function) tuples
    """
    results = {}
    
    for check_name, check_func in checks:
        try:
            passed = check_func(df)
            results[check_name] = "PASSED" if passed else "FAILED"
        except Exception as e:
            results[check_name] = f"ERROR: {str(e)}"
    
    return results

# Define checks
quality_checks = [
    ("No Nulls in ID", lambda df: df.filter(col("id").isNull()).count() == 0),
    ("No Duplicates", lambda df: df.count() == df.dropDuplicates().count()),
    ("Valid Date Range", lambda df: df.filter((col("date") >= "2024-01-01") & (col("date") <= "2024-12-31")).count() > 0),
    ("Non-Empty", lambda df: df.count() > 0)
]

# Run validation
validation_results = validate_data_quality(processed_data, quality_checks)

if all(v == "PASSED" for v in validation_results.values()):
    # Write to destination
    processed_data.write.format("delta").mode("overwrite").save(path)
else:
    # Alert and don't proceed
    print(f"Data quality checks failed: {validation_results}")
    raise Exception("Data validation failed")

8. Complete Fault-Tolerant Pipeline Template

import logging
from datetime import datetime

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class FaultTolerantPipeline:
    def __init__(self, spark, checkpoint_path):
        self.spark = spark
        self.checkpoint_path = checkpoint_path
        self.job_state = f"{checkpoint_path}/job_state"
    
    def read_with_retry(self, source_path, max_retries=3):
        return retry_with_backoff(
            lambda: self.spark.read.format("csv").load(source_path),
            max_retries=max_retries
        )
    
    def validate_and_transform(self, df):
        # Validation
        if df.count() == 0:
            raise ValueError("Input data is empty")
        
        # Transformation
        transformed = df.filter(col("status").isin(["valid", "pending"])) \
                        .withColumn("processed_at", current_timestamp())
        
        return transformed
    
    def write_with_idempotency(self, df, path):
        delta_table = DeltaTable.forPath(self.spark, path)
        
        delta_table.merge(
            df,
            "target.id = source.id"
        ).whenMatchedUpdateAll() \
         .whenNotMatchedInsertAll() \
         .execute()
        
        logger.info(f"Successfully wrote {df.count()} records to {path}")
    
    def run(self, source_path, target_path):
        try:
            logger.info("Pipeline started")
            
            # Read
            raw_data = self.read_with_retry(source_path)
            
            # Transform
            clean_data = self.validate_and_transform(raw_data)
            
            # Write
            self.write_with_idempotency(clean_data, target_path)
            
            logger.info("Pipeline completed successfully")
            return True
            
        except Exception as e:
            logger.error(f"Pipeline failed: {str(e)}")
            # Can trigger alerts, notifications here
            raise

# Usage
pipeline = FaultTolerantPipeline(spark, "/mnt/checkpoints")
pipeline.run("/mnt/raw/data", "/mnt/delta/processed")

---

Conclusion

This comprehensive guide covers the essential concepts and practical implementations required for Tiger Analytics interview success. Focus on understanding not just the “what” but also the “why” behind each technology choice. Real-world scenarios often involve trade-offs, so develop a mindset of evaluating solutions based on your specific use case, data volume, latency requirements, and cost constraints.

Key areas to master:

• PySpark fundamentals: Lazy evaluation, caching, transformations
• Big data optimization: Joins, partitioning, indexing
• Cloud platforms: Azure’s data ecosystem
• Data engineering patterns: Incremental loads, CDC, stream processing
• Architectural thinking: Fault tolerance, scalability, monitoring

Good luck with your interviews!