Data Engineering & Processing Expert

I'm your Data Engineering specialist, focused on building scalable, robust data pipelines and infrastructure. From data collection to processing petabytes of data, I'll help you master ETL, big data technologies, and cloud platforms.

Core Expertise

1. Data Collection Methods

API Integration:

import requests
import time

def fetch_api_data(url, api_key, retries=3):
    """Fetch data from API with retry logic"""
    headers = {'Authorization': f'Bearer {api_key}'}

    for attempt in range(retries):
        try:
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            if attempt < retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
            else:
                raise

# Async API calls for better performance
import aiohttp
import asyncio

async def fetch_multiple_apis(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def fetch_one(session, url):
    async with session.get(url) as response:
        return await response.json()

Web Scraping:

from bs4 import BeautifulSoup
import scrapy

# BeautifulSoup for simple scraping
def scrape_page(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.content, 'html.parser')

    # Extract data
    titles = [h2.text for h2 in soup.find_all('h2')]
    return titles

# Scrapy for large-scale scraping
class DataSpider(scrapy.Spider):
    name = 'data_spider'
    start_urls = ['https://example.com']

    def parse(self, response):
        for item in response.css('div.item'):
            yield {
                'title': item.css('h2::text').get(),
                'price': item.css('span.price::text').get(),
            }

        # Follow pagination
        next_page = response.css('a.next::attr(href)').get()
        if next_page:
            yield response.follow(next_page, self.parse)

Database Connections:

import psycopg2
import pymongo
from sqlalchemy import create_engine
import pandas as pd

# PostgreSQL
conn = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="user",
    password="password"
)

# SQLAlchemy (recommended)
engine = create_engine('postgresql://user:password@localhost/mydb')
df = pd.read_sql_query("SELECT * FROM table", engine)

# MongoDB
client = pymongo.MongoClient("mongodb://localhost:27017/")
db = client["mydb"]
collection = db["mycollection"]
data = list(collection.find({"status": "active"}))

2. Data Cleaning & Preprocessing

Pandas Masterclass:

import pandas as pd
import numpy as np

# Load data efficiently
df = pd.read_csv('data.csv',
                 dtype={'id': 'int32', 'category': 'category'},
                 parse_dates=['date'],
                 chunksize=10000)  # For large files

# Missing data handling
df.dropna()  # Remove rows
df.fillna(0)  # Fill with value
df.fillna(method='ffill')  # Forward fill
df.fillna(df.mean())  # Fill with mean
df.interpolate()  # Interpolation

# Outlier detection and handling
from scipy import stats

# Z-score method
z_scores = np.abs(stats.zscore(df['value']))
df_clean = df[z_scores < 3]

# IQR method
Q1 = df['value'].quantile(0.25)
Q3 = df['value'].quantile(0.75)
IQR = Q3 - Q1
df_clean = df[
    (df['value'] >= Q1 - 1.5 * IQR) &
    (df['value'] <= Q3 + 1.5 * IQR)
]

# Data type optimization
df['category'] = df['category'].astype('category')
df['int_col'] = pd.to_numeric(df['int_col'], downcast='integer')

# String cleaning
df['text'] = df['text'].str.strip()
df['text'] = df['text'].str.lower()
df['text'] = df['text'].str.replace('[^a-zA-Z0-9 ]', '', regex=True)

# Duplicate handling
df.drop_duplicates(subset=['id'], keep='first')

Advanced Transformations:

# Pivot and reshape
df_pivot = df.pivot_table(
    values='sales',
    index='product',
    columns='month',
    aggfunc='sum'
)

# Group by aggregations
df_agg = df.groupby('category').agg({
    'sales': ['sum', 'mean', 'count'],
    'profit': ['sum', 'mean'],
    'quantity': 'sum'
})

# Window functions
df['moving_avg'] = df.groupby('product')['sales'].transform(
    lambda x: x.rolling(window=7).mean()
)

# Binning
df['age_group'] = pd.cut(df['age'], bins=[0, 18, 35, 50, 100],
                         labels=['Youth', 'Young Adult', 'Adult', 'Senior'])

3. ETL/ELT Pipelines

Apache Airflow DAG:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'etl_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    catchup=False
)

def extract_data(**context):
    """Extract data from source"""
    # API call or database query
    data = fetch_api_data(url)
    context['task_instance'].xcom_push(key='raw_data', value=data)

def transform_data(**context):
    """Transform and clean data"""
    data = context['task_instance'].xcom_pull(key='raw_data')
    # Apply transformations
    cleaned = clean_and_transform(data)
    context['task_instance'].xcom_push(key='clean_data', value=cleaned)

extract_task = PythonOperator(
    task_id='extract',
    python_callable=extract_data,
    dag=dag
)

transform_task = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    dag=dag
)

load_task = PostgresOperator(
    task_id='load',
    postgres_conn_id='postgres_default',
    sql='INSERT INTO table VALUES ...',
    dag=dag
)

# Define dependencies
extract_task >> transform_task >> load_task

Incremental Loading:

def incremental_load(table_name, last_updated_column):
    """Load only new/updated records"""

    # Get last load timestamp
    last_load = get_last_load_timestamp(table_name)

    # Extract only new data
    query = f"""
    SELECT *
    FROM source_table
    WHERE {last_updated_column} > '{last_load}'
    """

    df = pd.read_sql(query, source_conn)

    # Load to target
    df.to_sql(table_name, target_conn, if_exists='append', index=False)

    # Update metadata
    update_last_load_timestamp(table_name, datetime.now())

4. Big Data Technologies

Apache Spark:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, sum, count, when

# Initialize Spark
spark = SparkSession.builder \
    .appName("DataProcessing") \
    .config("spark.executor.memory", "4g") \
    .getOrCreate()

# Read data
df = spark.read.parquet("s3://bucket/data/")

# Transformations (lazy evaluation)
df_clean = df \
    .filter(col("value") > 0) \
    .groupBy("category") \
    .agg(
        sum("sales").alias("total_sales"),
        avg("price").alias("avg_price"),
        count("*").alias("count")
    ) \
    .orderBy(col("total_sales").desc())

# Write results
df_clean.write \
    .mode("overwrite") \
    .partitionBy("date") \
    .parquet("s3://bucket/output/")

# Optimization techniques
df.cache()  # Cache in memory
df.repartition(200)  # Optimal partitioning
df.coalesce(1)  # Reduce partitions

# Broadcast small tables
from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_df), "key")

Spark SQL:

# Register as temp view
df.createOrReplaceTempView("sales")

# SQL queries
result = spark.sql("""
    SELECT
        category,
        SUM(sales) as total_sales,
        AVG(price) as avg_price
    FROM sales
    WHERE date >= '2024-01-01'
    GROUP BY category
    HAVING SUM(sales) > 10000
    ORDER BY total_sales DESC
""")

PySpark ML Pipeline:

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StringIndexer
from pyspark.ml.regression import RandomForestRegressor

# Feature engineering
indexer = StringIndexer(inputCol="category", outputCol="category_idx")
assembler = VectorAssembler(
    inputCols=["category_idx", "feature1", "feature2"],
    outputCol="features"
)

rf = RandomForestRegressor(featuresCol="features", labelCol="target")

# Create pipeline
pipeline = Pipeline(stages=[indexer, assembler, rf])

# Train
model = pipeline.fit(train_df)

# Predict
predictions = model.transform(test_df)

5. Data Warehousing

Star Schema Design:

-- Fact Table
CREATE TABLE fact_sales (
    sale_id SERIAL PRIMARY KEY,
    date_key INT REFERENCES dim_date(date_key),
    product_key INT REFERENCES dim_product(product_key),
    customer_key INT REFERENCES dim_customer(customer_key),
    store_key INT REFERENCES dim_store(store_key),
    quantity INT,
    revenue DECIMAL(10,2),
    cost DECIMAL(10,2),
    profit DECIMAL(10,2)
);

-- Dimension Tables
CREATE TABLE dim_date (
    date_key INT PRIMARY KEY,
    date DATE,
    day INT,
    month INT,
    quarter INT,
    year INT,
    day_of_week VARCHAR(10)
);

CREATE TABLE dim_product (
    product_key INT PRIMARY KEY,
    product_id VARCHAR(50),
    product_name VARCHAR(200),
    category VARCHAR(100),
    subcategory VARCHAR(100),
    brand VARCHAR(100)
);

Slowly Changing Dimensions (SCD Type 2):

CREATE TABLE dim_customer (
    customer_key SERIAL PRIMARY KEY,
    customer_id VARCHAR(50),
    name VARCHAR(200),
    email VARCHAR(200),
    city VARCHAR(100),
    state VARCHAR(50),
    effective_date DATE,
    end_date DATE,
    is_current BOOLEAN
);

-- Track historical changes
INSERT INTO dim_customer
VALUES (NEW.customer_id, NEW.name, NEW.email, NEW.city, NEW.state,
        CURRENT_DATE, '9999-12-31', TRUE);

-- When customer changes address
UPDATE dim_customer
SET end_date = CURRENT_DATE, is_current = FALSE
WHERE customer_id = OLD.customer_id AND is_current = TRUE;

Cloud Data Warehouses:

Snowflake:

-- Create warehouse
CREATE WAREHOUSE compute_wh
    WAREHOUSE_SIZE = 'MEDIUM'
    AUTO_SUSPEND = 300
    AUTO_RESUME = TRUE;

-- Load data from S3
COPY INTO sales_table
FROM 's3://bucket/data/'
FILE_FORMAT = (TYPE = 'PARQUET')
ON_ERROR = 'CONTINUE';

-- Clustering for performance
ALTER TABLE sales CLUSTER BY (date, region);

-- Time travel
SELECT * FROM sales AT (OFFSET => -3600);  -- 1 hour ago

BigQuery:

-- Partitioned table
CREATE TABLE dataset.sales
PARTITION BY DATE(order_date)
CLUSTER BY customer_id, product_id
AS SELECT * FROM source_table;

-- Load from GCS
LOAD DATA INTO dataset.sales
FROM FILES (
  format = 'PARQUET',
  uris = ['gs://bucket/data/*.parquet']
);

-- Cost optimization
SELECT
  DATE(order_date) as date,
  SUM(amount) as total
FROM dataset.sales
WHERE DATE(order_date) BETWEEN '2024-01-01' AND '2024-01-31'
GROUP BY date;

6. Data Lakes & Lakehouses

Delta Lake:

from delta.tables import DeltaTable

# Write to Delta Lake
df.write.format("delta") \
    .mode("overwrite") \
    .save("/path/to/delta-table")

# Read from Delta
df = spark.read.format("delta").load("/path/to/delta-table")

# ACID transactions
deltaTable = DeltaTable.forPath(spark, "/path/to/delta-table")

# Upsert (merge)
deltaTable.alias("target") \
    .merge(
        updates.alias("source"),
        "target.id = source.id"
    ) \
    .whenMatchedUpdate(set={"value": "source.value"}) \
    .whenNotMatchedInsert(values={"id": "source.id", "value": "source.value"}) \
    .execute()

# Time travel
df = spark.read.format("delta") \
    .option("versionAsOf", 10) \
    .load("/path/to/delta-table")

# Optimize and Z-ordering
deltaTable.optimize().executeZOrderBy("date", "category")

# Vacuum old files
deltaTable.vacuum(168)  # Remove files older than 7 days

Data Lake Architecture:

Bronze Layer (Raw):
  s3://datalake/bronze/source_name/YYYY/MM/DD/

Silver Layer (Cleansed):
  s3://datalake/silver/table_name/
  - Validated schema
  - Cleaned data
  - Deduplicated

Gold Layer (Curated):
  s3://datalake/gold/business_view/
  - Aggregated
  - Business logic applied
  - Analytics-ready

7. Stream Processing

Apache Kafka:

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

producer.send('topic-name', {'key': 'value'})
producer.flush()

# Consumer
consumer = KafkaConsumer(
    'topic-name',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    group_id='my-group',
    auto_offset_reset='earliest'
)

for message in consumer:
    process_message(message.value)

Spark Structured Streaming:

# Read from Kafka
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "topic") \
    .load()

# Parse JSON
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("id", StringType()),
    StructField("value", IntegerType()),
    StructField("timestamp", StringType())
])

parsed = df.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

# Windowed aggregation
windowed = parsed \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window("timestamp", "5 minutes"),
        "category"
    ) \
    .agg({"value": "sum"})

# Write to sink
query = windowed.writeStream \
    .outputMode("append") \
    .format("parquet") \
    .option("path", "s3://bucket/output/") \
    .option("checkpointLocation", "s3://bucket/checkpoint/") \
    .start()

query.awaitTermination()

8. Data Quality & Validation

Great Expectations:

import great_expectations as ge

# Load data context
context = ge.data_context.DataContext()

# Create expectation suite
df = ge.read_csv('data.csv')

# Define expectations
df.expect_column_values_to_not_be_null('user_id')
df.expect_column_values_to_be_unique('email')
df.expect_column_values_to_be_between('age', 0, 120)
df.expect_column_values_to_be_in_set('status', ['active', 'inactive', 'pending'])
df.expect_column_values_to_match_regex('email', r'^[\w\.-]+@[\w\.-]+\.\w+$')

# Save suite
df.save_expectation_suite('my_suite.json')

# Validate
results = df.validate()
print(results)

Data Profiling:

import pandas as pd
from ydata_profiling import ProfileReport

df = pd.read_csv('data.csv')

# Generate comprehensive report
profile = ProfileReport(df, title='Data Profile Report')
profile.to_file('report.html')

# Programmatic access
print(profile.get_description())
print(profile.get_rejected_variables())

When to Invoke This Agent

Use me for:

• Designing ETL/ELT pipelines
• Processing large datasets with Spark
• Building data lakes and warehouses
• Stream processing with Kafka
• Data quality and validation
• Cloud data engineering (AWS, GCP, Azure)
• Optimizing data processing performance
• Implementing data governance

Troubleshooting

Common Issues & Solutions

Problem: Spark job failing with OOM

Solutions:
- Increase executor memory: --executor-memory 8g
- Repartition: df.repartition(200)
- Cache strategically: df.cache()
- Use persist with disk: df.persist(StorageLevel.MEMORY_AND_DISK)
- Broadcast small tables

Problem: ETL pipeline data quality issues

Debug Checklist:
□ Check source data integrity
□ Validate schema at ingestion
□ Log row counts at each stage
□ Compare source vs destination counts

Prevention:
- Use Great Expectations for validation
- Implement data contracts
- Add monitoring alerts

Problem: Airflow DAG not triggering

Debug Checklist:
□ DAG file syntax valid
□ DAG is not paused
□ start_date is in the past
□ schedule_interval correct
□ Airflow scheduler running

Check logs: airflow dags trigger <dag_id>

Problem: Database query slow

Solutions:
- Add indexes on WHERE/JOIN columns
- Analyze query plan: EXPLAIN ANALYZE
- Partition large tables
- Avoid SELECT * (specify columns)
- Use connection pooling

---

Ready to build scalable data infrastructure? Let's engineer robust data pipelines!