ETL & Data Engineering Interview

import boto3
import pandas as pd
import re
from io import StringIO
from datetime import datetime

# ─── Step 1: Read CSV from S3 ───────────────────────────
s3 = boto3.client('s3')
response = s3.get_object(
    Bucket='my-data-bucket',
    Key='incoming/customers_20260323.csv'
)
csv_content = response['Body'].read().decode('utf-8')
df = pd.read_csv(StringIO(csv_content))

# ─── Step 2: Validate mobile numbers ───────────────────
# Regex: exactly 10 digits, nothing else
mobile_pattern = re.compile(r'^[0-9]{10}$')

def is_valid_mobile(value):
    """Return True if mobile is exactly 10 digits."""
    if pd.isna(value):          # Handle NULL/NaN
        return False
    value = str(value).strip()  # Remove whitespace
    return bool(mobile_pattern.match(value))

# Create a boolean mask
mask = df['mobile_number'].apply(is_valid_mobile)

# ─── Step 3: Split into valid and invalid ──────────────
valid_df   = df[mask].copy()
invalid_df = df[~mask].copy()

# Add error details to invalid records
invalid_df['error_reason'] = invalid_df['mobile_number'].apply(
    lambda x: 'NULL value' if pd.isna(x)
    else f'Invalid length: {len(str(x).strip())} digits'
      if str(x).strip().isdigit()
    else 'Contains non-numeric characters'
)
invalid_df['error_timestamp'] = datetime.now()

# ─── Step 4: Load valid records to database ────────────
from sqlalchemy import create_engine
engine = create_engine('mssql+pyodbc://user:pass@server/db')
valid_df.to_sql('customers', engine, if_exists='append', index=False)

# ─── Step 5: Write errors back to S3 ───────────────────
error_csv = invalid_df.to_csv(index=False)
s3.put_object(
    Bucket='my-data-bucket',
    Key=f'errors/customers_errors_{datetime.now():%Y%m%d}.csv',
    Body=error_csv
)

# ─── Step 6: Log summary ───────────────────────────────
print(f"Total: {len(df)} | Valid: {len(valid_df)} | Invalid: {len(invalid_df)}")
# Output: Total: 100 | Valid: 98 | Invalid: 2

-- Step 1: Create staging table (accepts everything)
CREATE TABLE stg_customers (
    customer_id     INT,
    customer_name   VARCHAR(100),
    mobile_number   VARCHAR(50),   -- VARCHAR, not INT (to catch "abc")
    email           VARCHAR(200),
    load_date       DATETIME DEFAULT GETDATE()
);

-- Step 2: Bulk load CSV into staging
BULK INSERT stg_customers
FROM '\\path\to\customers.csv'
WITH (FIELDTERMINATOR = ',', ROWTERMINATOR = '\n', FIRSTROW = 2);

-- Step 3: Move VALID records to production table
INSERT INTO customers (customer_id, customer_name, mobile_number, email)
SELECT customer_id, customer_name, mobile_number, email
FROM stg_customers
WHERE mobile_number IS NOT NULL
  AND LEN(mobile_number) = 10
  AND mobile_number NOT LIKE '%[^0-9]%';  -- no non-numeric chars

-- Step 4: Move INVALID records to error table
INSERT INTO error_log (customer_id, mobile_number, error_reason, error_date)
SELECT
    customer_id,
    mobile_number,
    CASE
        WHEN mobile_number IS NULL THEN 'NULL mobile number'
        WHEN mobile_number LIKE '%[^0-9]%' THEN 'Contains non-numeric characters'
        WHEN LEN(mobile_number) < 10 THEN 'Too short: ' + CAST(LEN(mobile_number) AS VARCHAR)
        WHEN LEN(mobile_number) > 10 THEN 'Too long: ' + CAST(LEN(mobile_number) AS VARCHAR)
    END,
    GETDATE()
FROM stg_customers
WHERE mobile_number IS NULL
   OR LEN(mobile_number) != 10
   OR mobile_number LIKE '%[^0-9]%';

-- Step 5: Clean up staging
TRUNCATE TABLE stg_customers;

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    name: str
    email: str
    mobile: str

# ─── GET: Read data ─────────────────────────────────────
@app.get("/users")
def get_all_users():
    return db.fetch_all_users()

@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = db.find_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# ─── POST: Create new resource ──────────────────────────
@app.post("/users", status_code=201)
def create_user(user: User):
    new_id = db.insert_user(user)
    return {"id": new_id, "message": "Created"}

# ─── PUT: Replace entire resource ───────────────────────
@app.put("/users/{user_id}")
def replace_user(user_id: int, user: User):
    db.update_user(user_id, user)  # overwrites ALL fields
    return {"message": "Replaced"}

# ─── PATCH: Update part of resource ─────────────────────
@app.patch("/users/{user_id}")
def update_email(user_id: int, email: str):
    db.update_field(user_id, "email", email)  # only email
    return {"message": "Updated"}

# ─── DELETE: Remove resource ────────────────────────────
@app.delete("/users/{user_id}")
def delete_user(user_id: int):
    db.remove_user(user_id)
    return {"message": "Deleted"}

-- Find employees earning more than their L1 manager
SELECT
    e.employee_code,
    e.employee_name    AS employee_name,
    e.salary           AS employee_salary,
    m.employee_name    AS manager_name,
    m.salary           AS manager_salary,
    e.salary - m.salary AS salary_difference
FROM employees e                    -- "e" = employee
INNER JOIN employees m              -- "m" = manager (same table!)
    ON e.l1_employee_code = m.employee_code
WHERE e.salary > m.salary
ORDER BY salary_difference DESC;

-- Include employees whose manager is not in the table
SELECT
    e.employee_name,
    e.salary,
    COALESCE(m.employee_name, 'No Manager') AS manager_name,
    m.salary AS manager_salary
FROM employees e
LEFT JOIN employees m
    ON e.l1_employee_code = m.employee_code
WHERE e.salary > m.salary
   OR m.employee_code IS NULL;  -- also show employees with no manager

# ─── Hashing (one-way, for passwords) ──────────────────
from passlib.hash import bcrypt

password = "user_password_123"
hashed = bcrypt.hash(password)
# "$2b$12$LJ3m4ys..." — cannot reverse this!

# Verify: hash the input again and compare
is_valid = bcrypt.verify("user_password_123", hashed)  # True

# ─── Encryption (two-way, for sensitive data) ──────────
from cryptography.fernet import Fernet

key = Fernet.generate_key()           # keep this secret!
cipher = Fernet(key)

# Encrypt
ssn = "123-45-6789"
encrypted = cipher.encrypt(ssn.encode())
# b'gAAAAABh...' — unreadable

# Decrypt (only with the same key)
original = cipher.decrypt(encrypted).decode()
# "123-45-6789" — got it back!

import pandas as pd
from sqlalchemy import create_engine
import boto3
from datetime import datetime

# Step 1: Extract data from database
engine = create_engine("mssql+pyodbc://...")
query = """
    SELECT customer_id, name, email, status
    FROM customers
    WHERE modified_date >= DATEADD(day, -1, GETDATE())
"""
df = pd.read_sql(query, engine)

# Step 2: Transform (select only outbound columns, mask PII)
outbound_df = df[['customer_id', 'name', 'status']]

# Step 3: Write to CSV and upload to S3 (for partner pickup)
filename = f"outbound/customers_{datetime.now():%Y%m%d}.csv"
csv_buffer = outbound_df.to_csv(index=False)

s3 = boto3.client('s3')
s3.put_object(
    Bucket='partner-data-exchange',
    Key=filename,
    Body=csv_buffer
)

# Step 4: Log the transfer
log_transfer(
    records=len(outbound_df),
    destination="partner-s3",
    filename=filename
)

@app.get("/api/v1/customers")
def get_customers(
    modified_since: str = None,   # "2026-03-22"
    limit: int = 100,
    offset: int = 0,
    api_key: str = Header(...)     # auth required
):
    validate_api_key(api_key)       # check partner has access

    query = db.query(Customer)
    if modified_since:
        query = query.filter(Customer.modified_date >= modified_since)

    results = query.offset(offset).limit(limit).all()

    return {
        "data": results,
        "count": len(results),
        "next_offset": offset + limit
    }

-- Step 1: Add a hash column to track outbound changes
ALTER TABLE customers ADD outbound_hash VARBINARY(32);

-- Step 2: Compute hash of ONLY the 5 outbound columns
UPDATE customers
SET outbound_hash = HASHBYTES(
    'SHA2_256',
    CONCAT(
        COALESCE(CAST(name AS VARCHAR(MAX)), '__NULL__'), '|',
        COALESCE(CAST(email AS VARCHAR(MAX)), '__NULL__'), '|',
        COALESCE(CAST(status AS VARCHAR(MAX)), '__NULL__'), '|',
        COALESCE(CAST(mobile AS VARCHAR(MAX)), '__NULL__'), '|',
        COALESCE(CAST(address AS VARCHAR(MAX)), '__NULL__')
    )
);

-- Step 3: On next ETL run, find rows where hash changed
SELECT c.*
FROM customers c
INNER JOIN outbound_snapshot prev
    ON c.customer_id = prev.customer_id
WHERE c.outbound_hash != prev.outbound_hash;

-- These are the ONLY rows to send outbound

-- Compare current table vs last snapshot
SELECT curr.*
FROM customers curr
INNER JOIN customers_snapshot prev
    ON curr.customer_id = prev.customer_id
WHERE
    -- Check ONLY the 5 outbound columns
    ISNULL(curr.name, '')    != ISNULL(prev.name, '')
    OR ISNULL(curr.email, '')   != ISNULL(prev.email, '')
    OR ISNULL(curr.status, '')  != ISNULL(prev.status, '')
    OR ISNULL(curr.mobile, '')  != ISNULL(prev.mobile, '')
    OR ISNULL(curr.address, '') != ISNULL(prev.address, '');

-- Note: Changes to col6-col20 are completely ignored

outbound_cols = ['name', 'email', 'status', 'mobile', 'address']
key_col = 'customer_id'

# Current data (from DB) and previous snapshot
current = pd.read_sql("SELECT * FROM customers", engine)
previous = pd.read_csv("last_snapshot.csv")

# Merge on key, compare only outbound columns
merged = current.merge(
    previous[[key_col] + outbound_cols],
    on=key_col,
    suffixes=('_curr', '_prev'),
    how='left'
)

# Find rows where ANY of the 5 outbound columns changed
changed_mask = False
for col in outbound_cols:
    changed_mask = changed_mask | (
        merged[f'{col}_curr'].fillna('__NULL__') !=
        merged[f'{col}_prev'].fillna('__NULL__')
    )

changed_rows = current[merged[changed_mask].index]
print(f"Rows to send outbound: {len(changed_rows)}")

-- ─── Local Temp Table (single #) ──────────────────────
-- Only THIS session/connection can see it
CREATE TABLE #my_staging (
    id      INT,
    name    VARCHAR(100),
    amount  DECIMAL(10,2)
);

INSERT INTO #my_staging VALUES (1, 'Alice', 5000.00);

-- Use it in your ETL logic
SELECT * FROM #my_staging;

-- Automatically dropped when you close the connection
-- (or manually: DROP TABLE #my_staging)

-- ─── Global Temp Table (double ##) ────────────────────
-- ALL sessions can see and use it
CREATE TABLE ##shared_lookup (
    code    VARCHAR(10),
    label   VARCHAR(100)
);

INSERT INTO ##shared_lookup VALUES ('US', 'United States');

-- Another user in a different session can do:
-- SELECT * FROM ##shared_lookup;  ← This works!

-- Dropped when the CREATING session ends AND
-- no other session is referencing it

import boto3
from datetime import datetime, timezone, timedelta

def delete_old_s3_files(
    bucket_name: str,
    prefix: str,
    days_old: int = 30,
    dry_run: bool = True
):
    """Delete S3 objects older than `days_old` days."""
    s3 = boto3.client('s3')
    cutoff_date = datetime.now(timezone.utc) - timedelta(days=days_old)

    deleted_count = 0
    total_size = 0

    # Paginator handles buckets with 1000+ objects
    paginator = s3.get_paginator('list_objects_v2')
    pages = paginator.paginate(Bucket=bucket_name, Prefix=prefix)

    for page in pages:
        if 'Contents' not in page:
            print("No objects found.")
            return

        for obj in page['Contents']:
            last_modified = obj['LastModified']  # already timezone-aware
            key = obj['Key']
            size = obj['Size']

            if last_modified < cutoff_date:
                age_days = (datetime.now(timezone.utc) - last_modified).days

                if dry_run:
                    print(f"[DRY RUN] Would delete: {key} "
                          f"(age: {age_days}d, size: {size:,}B)")
                else:
                    s3.delete_object(Bucket=bucket_name, Key=key)
                    print(f"[DELETED] {key} (age: {age_days}d)")

                deleted_count += 1
                total_size += size

    print(f"\nSummary: {deleted_count} files, {total_size / 1024 / 1024:.1f} MB")
    if dry_run:
        print("Run with dry_run=False to actually delete.")

# ─── Usage ──────────────────────────────────────────────
delete_old_s3_files(
    bucket_name='my-etl-bucket',
    prefix='raw-data/',        # only this "folder"
    days_old=30,
    dry_run=True               # set False to actually delete
)

# Collect keys to delete, then batch delete (up to 1000 at once)
keys_to_delete = [{'Key': key} for key in old_keys]

# Batch in groups of 1000
for i in range(0, len(keys_to_delete), 1000):
    batch = keys_to_delete[i:i+1000]
    s3.delete_objects(
        Bucket=bucket_name,
        Delete={'Objects': batch}
    )

aws s3api put-bucket-lifecycle-configuration \
  --bucket my-etl-bucket \
  --lifecycle-configuration '{
    "Rules": [{
      "ID": "DeleteOldRawData",
      "Filter": {"Prefix": "raw-data/"},
      "Status": "Enabled",
      "Expiration": {"Days": 30}
    }]
  }'

-- Step 1: Use CTE to assign row numbers within each duplicate group
WITH cte_duplicates AS (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY name, email, mobile   -- columns that define "duplicate"
            ORDER BY id                         -- keep the lowest id
        ) AS row_num
    FROM customers
)

-- Step 2: Delete all rows where row_num > 1 (the duplicates)
DELETE FROM cte_duplicates
WHERE row_num > 1;

-- First: SEE what will be deleted (don't actually delete)
WITH cte_duplicates AS (
    SELECT *, ROW_NUMBER() OVER (
        PARTITION BY name, email, mobile
        ORDER BY id
    ) AS row_num
    FROM customers
)
SELECT * FROM cte_duplicates WHERE row_num > 1;
-- ↑ This SHOWS duplicates. Replace SELECT with DELETE when satisfied.

WITH cte AS (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY name, email, mobile
            ORDER BY (SELECT NULL)   -- arbitrary order, just pick one
        ) AS rn
    FROM customers
)
DELETE FROM cte WHERE rn > 1;

CREATE TABLE etl_run_log (
    run_id              INT IDENTITY(1,1) PRIMARY KEY,
    job_name            VARCHAR(100),       -- "customer_import", "order_sync"
    source_system       VARCHAR(50),        -- "S3", "SFTP", "API"
    target_system       VARCHAR(50),        -- "SQL Server", "Data Warehouse"
    run_date            DATETIME DEFAULT GETDATE(),
    records_read        INT,
    records_written     INT,
    records_failed      INT,
    file_name           VARCHAR(500),
    file_size_mb        DECIMAL(10,2),
    duration_seconds    INT,
    status              VARCHAR(20),        -- 'SUCCESS', 'FAILED', 'PARTIAL'
    error_message       VARCHAR(MAX)
);

import time
from datetime import datetime

def run_etl_with_logging(job_name: str, source: str, target: str):
    start_time = time.time()
    status = "SUCCESS"
    error_msg = None
    records_read = 0
    records_written = 0
    records_failed = 0

    try:
        # ── Your actual ETL logic here ────────────
        df = extract_from_s3(source)
        records_read = len(df)

        valid, invalid = validate(df)
        records_failed = len(invalid)

        load_to_db(valid, target)
        records_written = len(valid)
        # ─────────────────────────────────────────

    except Exception as e:
        status = "FAILED"
        error_msg = str(e)

    finally:
        duration = int(time.time() - start_time)

        # Log to database
        db.execute("""
            INSERT INTO etl_run_log
            (job_name, source_system, target_system,
             records_read, records_written, records_failed,
             duration_seconds, status, error_message)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            job_name, source, target,
            records_read, records_written, records_failed,
            duration, status, error_msg
        ))

-- Daily transfer summary (last 30 days)
SELECT
    CAST(run_date AS DATE)   AS transfer_date,
    COUNT(*)                  AS total_jobs,
    SUM(records_read)         AS total_records_read,
    SUM(records_written)      AS total_records_written,
    SUM(records_failed)       AS total_records_failed,
    SUM(file_size_mb)         AS total_data_mb,
    AVG(duration_seconds)     AS avg_duration_sec,
    SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END) AS failed_jobs
FROM etl_run_log
WHERE run_date >= DATEADD(day, -30, GETDATE())
GROUP BY CAST(run_date AS DATE)
ORDER BY transfer_date;

-- Monthly summary (for management)
SELECT
    FORMAT(run_date, 'yyyy-MM')  AS month,
    job_name,
    SUM(records_written)          AS total_records,
    SUM(file_size_mb)             AS total_gb,
    ROUND(
        100.0 * SUM(CASE WHEN status = 'SUCCESS' THEN 1 ELSE 0 END)
        / COUNT(*), 1
    )                             AS success_rate_pct
FROM etl_run_log
GROUP BY FORMAT(run_date, 'yyyy-MM'), job_name
ORDER BY month DESC, total_records DESC;