DockerAndSQLStuff.txt

# docker-compose.yml
# Smart City IoT Analytics Pipeline Infrastructure

version: '3.8'

services:
  # Apache Spark Master
  spark-master:
    image: bitnami/spark:3.4.0
    container_name: spark-master
    environment:
      - SPARK_MODE=master
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
      - SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED=no
      - SPARK_SSL_ENABLED=no
      - SPARK_MASTER_OPTS=-Dspark.deploy.defaultCores=2
    ports:
      - "8080:8080"  # Spark Master Web UI
      - "7077:7077"  # Spark Master Port
    volumes:
      - ./data:/opt/bitnami/spark/data
      - ./config:/opt/bitnami/spark/conf
      - ./notebooks:/opt/bitnami/spark/notebooks
    networks:
      - spark-network

  # Apache Spark Worker 1
  spark-worker-1:
    image: bitnami/spark:3.4.0
    container_name: spark-worker-1
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
      - SPARK_WORKER_MEMORY=2g
      - SPARK_WORKER_CORES=2
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
      - SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED=no
      - SPARK_SSL_ENABLED=no
    ports:
      - "8081:8081"  # Worker Web UI
    volumes:
      - ./data:/opt/bitnami/spark/data
      - ./config:/opt/bitnami/spark/conf
    depends_on:
      - spark-master
    networks:
      - spark-network

  # Apache Spark Worker 2
  spark-worker-2:
    image: bitnami/spark:3.4.0
    container_name: spark-worker-2
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
      - SPARK_WORKER_MEMORY=2g
      - SPARK_WORKER_CORES=2
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
      - SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED=no
      - SPARK_SSL_ENABLED=no
    ports:
      - "8082:8081"  # Worker Web UI
    volumes:
      - ./data:/opt/bitnami/spark/data
      - ./config:/opt/bitnami/spark/conf
    depends_on:
      - spark-master
    networks:
      - spark-network

  # PostgreSQL Database
  postgres:
    image: postgres:15-alpine
    container_name: postgres-smartcity
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=smartcity
      - POSTGRES_INITDB_ARGS=--encoding=UTF-8 --lc-collate=C --lc-ctype=C
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./sql/init.sql:/docker-entrypoint-initdb.d/init.sql
    networks:
      - spark-network
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d smartcity"]
      interval: 10s
      timeout: 5s
      retries: 5

  # Jupyter Notebook Server
  jupyter:
    build:
      context: .
      dockerfile: jupyter.Dockerfile
    container_name: jupyter-smartcity
    environment:
      - JUPYTER_ENABLE_LAB=yes
      - JUPYTER_TOKEN=smartcity123
      - SPARK_MASTER=spark://spark-master:7077
    ports:
      - "8888:8888"
    volumes:
      - ./notebooks:/home/jovyan/work/notebooks
      - ./data:/home/jovyan/work/data
      - ./src:/home/jovyan/work/src
      - ./config:/home/jovyan/work/config
    depends_on:
      - spark-master
      - postgres
    networks:
      - spark-network

  # Grafana Dashboard (Optional - for advanced visualization)
  grafana:
    image: grafana/grafana:latest
    container_name: grafana-smartcity
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin123
      - GF_INSTALL_PLUGINS=grafana-worldmap-panel,grafana-clock-panel
    ports:
      - "3000:3000"
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/provisioning:/etc/grafana/provisioning
    depends_on:
      - postgres
    networks:
      - spark-network

  # Redis (for caching and session management)
  redis:
    image: redis:alpine
    container_name: redis-smartcity
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    networks:
      - spark-network

  # Adminer (Database Management Tool)
  adminer:
    image: adminer
    container_name: adminer-smartcity
    ports:
      - "8080:8080"
    depends_on:
      - postgres
    networks:
      - spark-network

volumes:
  postgres_data:
    driver: local
  grafana_data:
    driver: local
  redis_data:
    driver: local

networks:
  spark-network:
    driver: bridge

---

# jupyter.Dockerfile
# Custom Jupyter image with PySpark and required dependencies

FROM jupyter/pyspark-notebook:latest

USER root

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    vim \
    git \
    && rm -rf /var/lib/apt/lists/*

# Install PostgreSQL client
RUN apt-get update && apt-get install -y postgresql-client && rm -rf /var/lib/apt/lists/*

USER $NB_UID

# Install Python packages
COPY requirements.txt /tmp/
RUN pip install -r /tmp/requirements.txt

# Install additional Jupyter extensions
RUN pip install \
    jupyterlab-git \
    jupyter-resource-usage \
    ipywidgets

# Set up Spark configuration
ENV SPARK_OPTS="--driver-java-options=-Xms2g --driver-java-options=-Xmx4g"

# Copy custom Spark configuration
COPY config/spark-defaults.conf $SPARK_HOME/conf/

# Set working directory
WORKDIR /home/jovyan/work

---

# config/spark-defaults.conf
# Spark Configuration for Smart City IoT Pipeline

# Application Properties
spark.app.name                 SmartCityIoTPipeline
spark.master                   spark://spark-master:7077

# Driver Configuration
spark.driver.memory            4g
spark.driver.cores             2
spark.driver.maxResultSize     2g

# Executor Configuration  
spark.executor.memory          2g
spark.executor.cores           2
spark.executor.instances       2

# SQL and DataFrame Configuration
spark.sql.adaptive.enabled                    true
spark.sql.adaptive.coalescePartitions.enabled true
spark.sql.adaptive.skewJoin.enabled           true
spark.sql.execution.arrow.pyspark.enabled     true

# Serialization
spark.serializer                org.apache.spark.serializer.KryoSerializer
spark.kryoserializer.buffer.max 1024m

# Shuffle Configuration
spark.sql.shuffle.partitions    200
spark.shuffle.compress          true
spark.shuffle.spill.compress    true

# I/O Configuration
spark.sql.files.maxPartitionBytes    134217728  # 128MB
spark.sql.files.openCostInBytes      4194304    # 4MB

# PostgreSQL JDBC Configuration
spark.jars.packages              org.postgresql:postgresql:42.5.0

# Checkpoint Configuration
spark.sql.streaming.checkpointLocation    /opt/bitnami/spark/checkpoints

# History Server
spark.eventLog.enabled          true
spark.eventLog.dir              /opt/bitnami/spark/logs

# Performance Tuning
spark.sql.adaptive.advisoryPartitionSizeInBytes  134217728
spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin  0.2

---

# sql/init.sql
-- PostgreSQL Initialization Script for Smart City IoT Database

-- Create database schema
CREATE SCHEMA IF NOT EXISTS smartcity;
SET search_path TO smartcity;

-- Create sensor types lookup table
CREATE TABLE sensor_types (
    sensor_type_id SERIAL PRIMARY KEY,
    type_name VARCHAR(50) NOT NULL UNIQUE,
    description TEXT,
    measurement_unit VARCHAR(20),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Insert sensor types
INSERT INTO sensor_types (type_name, description, measurement_unit) VALUES
('traffic', 'Traffic flow and speed sensors', 'vehicles/hour'),
('air_quality', 'Air pollution monitoring sensors', 'μg/m³'),
('weather', 'Weather monitoring stations', 'various'),
('energy', 'Energy consumption meters', 'kWh');

-- Create city zones table
CREATE TABLE zones (
    zone_id VARCHAR(20) PRIMARY KEY,
    zone_name VARCHAR(100) NOT NULL,
    zone_type VARCHAR(50) NOT NULL,
    lat_min DECIMAL(10,8) NOT NULL,
    lat_max DECIMAL(10,8) NOT NULL,
    lon_min DECIMAL(11,8) NOT NULL,
    lon_max DECIMAL(11,8) NOT NULL,
    population INTEGER,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create sensors table
CREATE TABLE sensors (
    sensor_id VARCHAR(50) PRIMARY KEY,
    sensor_type_id INTEGER REFERENCES sensor_types(sensor_type_id),
    location_lat DECIMAL(10,8) NOT NULL,
    location_lon DECIMAL(11,8) NOT NULL,
    zone_id VARCHAR(20) REFERENCES zones(zone_id),
    installation_date DATE,
    last_maintenance DATE,
    status VARCHAR(20) DEFAULT 'active',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create main sensor readings table (fact table)
CREATE TABLE sensor_readings (
    reading_id BIGSERIAL PRIMARY KEY,
    sensor_id VARCHAR(50) REFERENCES sensors(sensor_id),
    timestamp TIMESTAMP NOT NULL,
    measurement_type VARCHAR(50) NOT NULL,
    value DECIMAL(15,6),
    unit VARCHAR(20),
    quality_score DECIMAL(3,2) DEFAULT 1.0,
    anomaly_flag BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create specific tables for each sensor type

-- Traffic sensor readings
CREATE TABLE traffic_readings (
    reading_id BIGSERIAL PRIMARY KEY,
    sensor_id VARCHAR(50) REFERENCES sensors(sensor_id),
    timestamp TIMESTAMP NOT NULL,
    vehicle_count INTEGER,
    avg_speed DECIMAL(5,2),
    congestion_level VARCHAR(20),
    road_type VARCHAR(50),
    quality_score DECIMAL(3,2) DEFAULT 1.0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Air quality readings
CREATE TABLE air_quality_readings (
    reading_id BIGSERIAL PRIMARY KEY,
    sensor_id VARCHAR(50) REFERENCES sensors(sensor_id),
    timestamp TIMESTAMP NOT NULL,
    pm25 DECIMAL(8,3),
    pm10 DECIMAL(8,3),
    no2 DECIMAL(8,3),
    co DECIMAL(8,4),
    temperature DECIMAL(5,2),
    humidity DECIMAL(5,2),
    quality_score DECIMAL(3,2) DEFAULT 1.0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Weather readings
CREATE TABLE weather_readings (
    reading_id BIGSERIAL PRIMARY KEY,
    station_id VARCHAR(50) REFERENCES sensors(sensor_id),
    timestamp TIMESTAMP NOT NULL,
    temperature DECIMAL(5,2),
    humidity DECIMAL(5,2),
    wind_speed DECIMAL(5,2),
    wind_direction DECIMAL(5,2),
    precipitation DECIMAL(6,3),
    pressure DECIMAL(7,2),
    quality_score DECIMAL(3,2) DEFAULT 1.0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Energy consumption readings
CREATE TABLE energy_readings (
    reading_id BIGSERIAL PRIMARY KEY,
    meter_id VARCHAR(50) REFERENCES sensors(sensor_id),
    timestamp TIMESTAMP NOT NULL,
    building_type VARCHAR(50),
    power_consumption DECIMAL(10,3),
    voltage DECIMAL(6,2),
    current DECIMAL(8,3),
    power_factor DECIMAL(4,3),
    quality_score DECIMAL(3,2) DEFAULT 1.0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create aggregated tables for common queries

-- Hourly aggregations
CREATE TABLE hourly_aggregates (
    aggregate_id BIGSERIAL PRIMARY KEY,
    zone_id VARCHAR(20) REFERENCES zones(zone_id),
    sensor_type VARCHAR(50),
    hour_timestamp TIMESTAMP NOT NULL,
    metric_name VARCHAR(100),
    avg_value DECIMAL(15,6),
    min_value DECIMAL(15,6),
    max_value DECIMAL(15,6),
    count_readings INTEGER,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Daily aggregations
CREATE TABLE daily_aggregates (
    aggregate_id BIGSERIAL PRIMARY KEY,
    zone_id VARCHAR(20) REFERENCES zones(zone_id),
    sensor_type VARCHAR(50),
    date_day DATE NOT NULL,
    metric_name VARCHAR(100),
    avg_value DECIMAL(15,6),
    min_value DECIMAL(15,6),
    max_value DECIMAL(15,6),
    count_readings INTEGER,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create indexes for performance
CREATE INDEX idx_sensor_readings_timestamp ON sensor_readings(timestamp);
CREATE INDEX idx_sensor_readings_sensor_id ON sensor_readings(sensor_id);
CREATE INDEX idx_traffic_readings_timestamp ON traffic_readings(timestamp);
CREATE INDEX idx_air_quality_readings_timestamp ON air_quality_readings(timestamp);
CREATE INDEX idx_weather_readings_timestamp ON weather_readings(timestamp);
CREATE INDEX idx_energy_readings_timestamp ON energy_readings(timestamp);

-- Create composite indexes for common query patterns
CREATE INDEX idx_sensor_readings_sensor_timestamp ON sensor_readings(sensor_id, timestamp);
CREATE INDEX idx_traffic_readings_sensor_timestamp ON traffic_readings(sensor_id, timestamp);
CREATE INDEX idx_hourly_aggregates_zone_type_time ON hourly_aggregates(zone_id, sensor_type, hour_timestamp);

-- Create views for common analytical queries

-- Current sensor status view
CREATE VIEW current_sensor_status AS
SELECT 
    s.sensor_id,
    st.type_name as sensor_type,
    s.location_lat,
    s.location_lon,
    s.zone_id,
    z.zone_name,
    z.zone_type,
    s.status,
    CASE 
        WHEN sr.last_reading < NOW() - INTERVAL '1 hour' THEN 'stale'
        WHEN sr.last_reading < NOW() - INTERVAL '6 hours' THEN 'warning'
        ELSE 'active'
    END as data_status,
    sr.last_reading
FROM sensors s
JOIN sensor_types st ON s.sensor_type_id = st.sensor_type_id
LEFT JOIN zones z ON s.zone_id = z.zone_id
LEFT JOIN (
    SELECT 
        sensor_id,
        MAX(timestamp) as last_reading
    FROM sensor_readings
    GROUP BY sensor_id
) sr ON s.sensor_id = sr.sensor_id;

-- Zone summary view
CREATE VIEW zone_summary AS
SELECT 
    z.zone_id,
    z.zone_name,
    z.zone_type,
    z.population,
    COUNT(DISTINCT s.sensor_id) as total_sensors,
    COUNT(DISTINCT CASE WHEN s.status = 'active' THEN s.sensor_id END) as active_sensors,
    COUNT(DISTINCT st.type_name) as sensor_types_count
FROM zones z
LEFT JOIN sensors s ON z.zone_id = s.zone_id
LEFT JOIN sensor_types st ON s.sensor_type_id = st.sensor_type_id
GROUP BY z.zone_id, z.zone_name, z.zone_type, z.population;

-- Grant permissions for application user
CREATE USER spark_user WITH PASSWORD 'spark_password';
GRANT USAGE ON SCHEMA smartcity TO spark_user;
GRANT SELECT, INSERT, UPDATE ON ALL TABLES IN SCHEMA smartcity TO spark_user;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA smartcity TO spark_user;

-- Add comments for documentation
COMMENT ON TABLE sensor_readings IS 'Main fact table for all sensor measurements';
COMMENT ON TABLE zones IS 'City zones and districts definition';
COMMENT ON TABLE sensors IS 'Sensor registry with locations and metadata';
COMMENT ON VIEW current_sensor_status IS 'Real-time view of sensor operational status';

-- Create function for data quality scoring
CREATE OR REPLACE FUNCTION calculate_data_quality_score(
    missing_values INTEGER,
    total_values INTEGER,
    outlier_count INTEGER
) RETURNS DECIMAL(3,2) AS $$
BEGIN
    -- Simple data quality scoring algorithm
    -- 1.0 = perfect quality, 0.0 = poor quality
    IF total_values = 0 THEN
        RETURN 0.0;
    END IF;
    
    RETURN GREATEST(0.0, 
        1.0 - (missing_values::DECIMAL / total_values) * 0.5 
            - (outlier_count::DECIMAL / total_values) * 0.3
    );
END;
$$ LANGUAGE plpgsql;

COMMENT ON FUNCTION calculate_data_quality_score IS 'Calculate data quality score based on missing values and outliers';

---

# requirements.txt
# Python dependencies for Smart City IoT Pipeline

# Core PySpark and data processing
pyspark==3.4.0
pandas==1.5.3
numpy==1.24.3
pyarrow==11.0.0

# Database connectivity
psycopg2-binary==2.9.5
sqlalchemy==2.0.7

# Data visualization
matplotlib==3.6.3
seaborn==0.12.2
plotly==5.13.1

# Dashboard and web interface
streamlit==1.20.0
dash==2.8.1
bokeh==3.1.0

# Machine learning
scikit-learn==1.2.2
scipy==1.10.1

# Time series analysis
statsmodels==0.13.5

# Geospatial analysis
geopandas==0.12.2
folium==0.14.0

# Utilities
requests==2.28.2
python-dotenv==1.0.0
click==8.1.3
tqdm==4.65.0

# Development and testing
pytest==7.2.2
black==23.1.0
flake8==6.0.0
jupyter==1.0.0

# Optional: For advanced features
redis==4.5.3
kafka-python==2.0.2

---

# .env
# Environment variables for Smart City IoT Pipeline

# Database Configuration
POSTGRES_HOST=postgres
POSTGRES_PORT=5432
POSTGRES_DB=smartcity
POSTGRES_USER=postgres
POSTGRES_PASSWORD=password
POSTGRES_SCHEMA=smartcity

# Spark Configuration
SPARK_MASTER_URL=spark://spark-master:7077
SPARK_DRIVER_MEMORY=4g
SPARK_EXECUTOR_MEMORY=2g

# Application Configuration
APP_ENV=development
DEBUG=true
LOG_LEVEL=INFO

# Dashboard Configuration
STREAMLIT_PORT=8501
GRAFANA_PORT=3000
GRAFANA_ADMIN_PASSWORD=admin123

# Redis Configuration
REDIS_HOST=redis
REDIS_PORT=6379

# Data Paths
DATA_DIR=/opt/bitnami/spark/data
CHECKPOINT_DIR=/opt/bitnami/spark/checkpoints