AWS EKS Enterprise Deployment: Real-Time Data Streaming Platform – 1 Million Events/Sec

┌──────────────────────────────────────────────────────────────────┐
│                  AWS EKS Cluster (us-west-2)                     │
│              benchmark-high-infra (k8s 1.31)                     │
├──────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────────────────┐   ┌──────────────────┐   ┌──────────────┐ │
│  │   PRODUCER      │──▶│     PULSAR       │──▶│    FLINK     │ │
│  │  c5.4xlarge     │   │  i7i.8xlarge     │   │ c5.4xlarge   │ │
│  │                 │   │                  │   │              │ │
│  │ 4 nodes         │   │ ZK + 6 Brokers   │   │ JM + 6 TMs   │ │
│  │ Java/AVRO       │   │ NVMe Storage     │   │ 1M evt/sec   │ │
│  │ 250K evt/sec    │   │ 3.6TB NVMe       │   │ Checkpoints  │ │
│  │ 100K devices    │   │ Ultra-low lat    │   │ Aggregation  │ │
│  └─────────────────┘   └──────────────────┘   └──────┬───────┘ │
│                                                        │         │
│                         ┌──────────────────────────────┘         │
│                         ▼                                        │
│                  ┌──────────────────┐                           │
│                  │   CLICKHOUSE     │                           │
│                  │  r6id.4xlarge    │                           │
│                  │                  │                           │
│                  │  6 Data Nodes    │                           │
│                  │  1 Query Node    │                           │
│                  │  NVMe + EBS      │                           │
│                  │  10K+ queries/s  │                           │
│                  └──────────────────┘                           │
│                                                                  │
│  Supporting: VPC, Multi-AZ, S3, ECR, IAM, Auto-scaling         │
└──────────────────────────────────────────────────────────────────┘

# Install required tools
brew install awscli terraform kubectl helm

# Configure AWS with admin-level access
aws configure
# Enter credentials for production account

# Verify versions
terraform --version  # >= 1.6.0
kubectl version      # >= 1.28.0
helm version         # >= 3.12.0

git clone https://github.com/hyperscaledesignhub/RealtimeDataPlatform.git
cd RealtimeDataPlatform/realtime-platform-1million-events

# Review configuration
cat terraform.tfvars

realtime-platform-1million-events/
├── terraform/                # Enterprise AWS infrastructure
├── producer-load/            # High-volume event generation
├── pulsar-load/              # Apache Pulsar (NVMe-backed)
├── flink-load/               # Apache Flink enterprise processing
├── clickhouse-load/          # ClickHouse analytics cluster
└── monitoring/               # Enterprise monitoring stack

# terraform.tfvars
cluster_name = "benchmark-high-infra"
aws_region = "us-west-2"
environment = "production"

# High-performance node groups
producer_desired_size = 4          # c5.4xlarge
pulsar_zookeeper_desired_size = 3  # t3.medium
pulsar_broker_desired_size = 6     # i7i.8xlarge (NVMe)
flink_taskmanager_desired_size = 6 # c5.4xlarge
clickhouse_desired_size = 6        # r6id.4xlarge (NVMe)

# Enable all services
enable_flink = true
enable_pulsar = true
enable_clickhouse = true
enable_general_nodes = true

# Initialize Terraform
terraform init

# Review infrastructure plan (~$24K-33K/month)
terraform plan

# Deploy infrastructure (takes ~20-25 minutes)
terraform apply -auto-approve

aws eks update-kubeconfig --region us-west-2 --name benchmark-high-infra

# Verify cluster
kubectl get nodes
# Should see ~30 nodes across all groups

cd pulsar-load

# Deploy Pulsar with NVMe storage
./deploy.sh

# Monitor deployment (~10-15 minutes for all components)
kubectl get pods -n pulsar -w

# Check all components are running
kubectl get pods -n pulsar

# Test Pulsar functionality
kubectl exec -n pulsar pulsar-broker-0 -- \
  bin/pulsar-admin topics create persistent://public/default/test-topic

# Verify topic creation
kubectl exec -n pulsar pulsar-broker-0 -- \
  bin/pulsar-admin topics list public/default

cd ../clickhouse-load

# Install ClickHouse operator and enterprise cluster
./00-install-clickhouse.sh

# Wait for ClickHouse cluster (~5-8 minutes)
kubectl get pods -n clickhouse -w

# Create enterprise database schema
./00-create-schema-all-replicas.sh

-- High-performance sensor data table using AVRO schema
CREATE TABLE IF NOT EXISTS benchmark.sensors_local ON CLUSTER iot_cluster (
    sensorId Int32,
    sensorType Int32,
    temperature Float64,
    humidity Float64,
    pressure Float64,
    batteryLevel Float64,
    status Int32,
    timestamp DateTime64(3),
    event_time DateTime64(3) DEFAULT now64()
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{cluster}/sensors_local', '{replica}')
PARTITION BY toYYYYMM(timestamp)
ORDER BY (sensorId, timestamp)
SETTINGS index_granularity = 8192;

# Connect to ClickHouse cluster
kubectl exec -it -n clickhouse chi-iot-cluster-repl-iot-cluster-0-0-0 -- clickhouse-client

# Test cluster connectivity
SELECT * FROM system.clusters WHERE cluster = 'iot_cluster';

# Exit with Ctrl+D

cd ../flink-load

# Build and push enterprise Flink image to ECR
./build-and-push.sh

# Deploy Flink enterprise cluster
./deploy.sh

# Submit high-throughput Flink job
kubectl apply -f flink-job-deployment.yaml

# Monitor Flink deployment (~3-5 minutes)
kubectl get pods -n flink-benchmark -w

// Enterprise-grade stream processing using SensorData AVRO schema
DataStream<SensorRecord> sensorStream = env.fromSource(
    pulsarSource,
    WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)),
    "Pulsar Enterprise IoT Source"
);

// High-throughput processing with 1-minute windows
sensorStream
    .keyBy(record -> record.getSensorId())
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new EnterpriseAggregator())
    .addSink(new ClickHouseJDBCSink(clickhouseUrl));

cd ../producer-load

# Build and deploy enterprise producer
./deploy.sh

# Scale to generate 1M events/sec (4 nodes × 250K each)
kubectl scale deployment iot-producer -n iot-pipeline --replicas=100

# Monitor producer performance
kubectl get pods -n iot-pipeline -l app=iot-producer

# Monitor producer throughput
kubectl logs -n iot-pipeline -l app=iot-producer --tail=20 | grep "Events produced"

# Check Pulsar message ingestion rate
kubectl exec -n pulsar pulsar-broker-0 -- \
  bin/pulsar-admin topics stats persistent://public/default/iot-sensor-data

# Verify Flink processing rate
kubectl logs -n flink-benchmark deployment/iot-flink-job --tail=20

# Query ClickHouse for ingestion rate
kubectl exec -n clickhouse chi-iot-cluster-repl-iot-cluster-0-0-0 -- \
  clickhouse-client --query "
    SELECT 
        toStartOfMinute(timestamp) as minute,
        COUNT(*) as events_per_minute
    FROM benchmark.sensors_local 
    WHERE timestamp >= now() - INTERVAL 5 MINUTE
    GROUP BY minute 
    ORDER BY minute DESC"

✅ Producer: 1,000,000+ events/sec generation
✅ Pulsar: Ultra-low latency message ingestion (~1ms)
✅ Flink: Real-time processing with exactly-once guarantees
✅ ClickHouse: High-speed data ingestion and sub-second queries
✅ End-to-end latency: < 2 seconds (p99)

# Set up secure port forwarding
kubectl port-forward -n pulsar svc/grafana 3000:3000 &

# Open enterprise dashboard
open http://localhost:3000
# Login: admin/admin

-- Connect to ClickHouse enterprise cluster
kubectl exec -it -n clickhouse chi-iot-cluster-repl-iot-cluster-0-0-0 -- clickhouse-client

-- Enterprise-scale analytics using our SensorData AVRO schema
USE benchmark;

-- Real-time throughput monitoring
SELECT 
    toStartOfMinute(timestamp) as minute,
    COUNT(*) as events_per_minute,
    COUNT(DISTINCT sensorId) as unique_sensors,
    AVG(temperature) as avg_temp,
    AVG(batteryLevel) as avg_battery
FROM sensors_local
WHERE timestamp >= now() - INTERVAL 1 HOUR
GROUP BY minute
ORDER BY minute DESC
LIMIT 60;

-- Enterprise anomaly detection
SELECT 
    sensorId,
    sensorType,
    temperature,
    batteryLevel,
    status,
    timestamp
FROM sensors_local
WHERE (temperature > 40.0 OR batteryLevel < 15.0 OR status != 1)
  AND timestamp >= now() - INTERVAL 10 MINUTE
ORDER BY timestamp DESC
LIMIT 100;

-- High-performance aggregations across millions of records
SELECT 
    sensorType,
    COUNT(*) as total_readings,
    AVG(temperature) as avg_temp,
    percentile(0.95)(temperature) as p95_temp,
    AVG(humidity) as avg_humidity,
    MIN(batteryLevel) as min_battery,
    MAX(batteryLevel) as max_battery
FROM sensors_local
WHERE timestamp >= today() - INTERVAL 1 DAY
GROUP BY sensorType
ORDER BY total_readings DESC;

-- Enterprise time-series analysis
SELECT 
    toStartOfHour(timestamp) as hour,
    sensorType,
    COUNT(*) as hourly_count,
    AVG(temperature) as avg_temp,
    stddevPop(temperature) as temp_stddev
FROM sensors_local
WHERE timestamp >= now() - INTERVAL 24 HOUR
GROUP BY hour, sensorType
ORDER BY hour DESC, sensorType;

# Check node resource utilization
kubectl top nodes | sort -k3 -nr

# Identify resource bottlenecks
kubectl describe nodes | grep -A5 "Allocated resources"

# Scale TaskManagers for higher throughput
kubectl scale deployment flink-taskmanager -n flink-benchmark --replicas=12

# Monitor Flink backpressure
kubectl exec -n flink-benchmark <jobmanager-pod> -- \
  flink list -r

# Check NVMe disk performance
kubectl exec -n pulsar pulsar-broker-0 -- \
  iostat -x 1 5

# Monitor ClickHouse storage usage
kubectl exec -n clickhouse chi-iot-cluster-repl-iot-cluster-0-0-0 -- \
  clickhouse-client --query "
    SELECT 
        name,
        total_space,
        free_space,
        (total_space - free_space) / total_space * 100 as usage_percent
    FROM system.disks"

# Check inter-pod network latency
kubectl exec -n pulsar pulsar-broker-0 -- \
  ping -c 5 flink-jobmanager.flink-benchmark.svc.cluster.local

# Monitor network bandwidth
kubectl exec -n flink-benchmark <taskmanager-pod> -- \
  iftop -t -s 10

# Graceful shutdown of applications
kubectl delete namespace iot-pipeline flink-benchmark

# Backup critical data before destroying infrastructure
./backup-clickhouse.sh
./backup-flink-savepoints.sh

# Destroy AWS infrastructure
terraform destroy
# Type 'yes' when prompted

# Verify all resources are cleaned up
aws ec2 describe-instances --region us-west-2 \
  --filters "Name=tag:kubernetes.io/cluster/benchmark-high-infra,Values=owned"

# Purchase 3-year reserved instances for 26% savings
# Target instances: i7i.8xlarge, r6id.4xlarge, c5.4xlarge

# AWS Console → EC2 → Reserved Instances → Purchase
# - Term: 3 years
# - Payment: All upfront (max discount)
# - Instance type: i7i.8xlarge, r6id.4xlarge
# - Quantity: Match your desired_size

# Savings: $33,016 → $24,592/month (26% off)

# Automated EBS snapshots
aws backup create-backup-plan --backup-plan-name daily-snapshots

# ClickHouse enterprise backups to S3
clickhouse-backup create
clickhouse-backup upload

# Flink savepoints for exactly-once recovery
kubectl exec -n flink-benchmark <jm-pod> -- \
  flink savepoint <job-id> s3://benchmark-high-infra-state/savepoints

# CloudWatch Alarms for enterprise monitoring
- CPU > 80% sustained for 5 minutes
- Disk usage > 85%
- Pod crash loops > 3 in 10 minutes
- Flink checkpoint failures
- Pulsar consumer lag > 1M messages
- ClickHouse replication lag > 5 minutes

# Deploy identical stack in secondary region
aws_region = "us-east-1"
cluster_name = "benchmark-high-infra-dr"

# Use Pulsar geo-replication
bin/pulsar-admin namespaces set-clusters public/default \
  --clusters us-west-2,us-east-1

# ClickHouse cross-region replication
CREATE TABLE benchmark.sensors_replicated
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{cluster}/sensors', '{replica}')
...

# Set up AWS Cost Explorer with enterprise tags
# Tag all resources:
# - Environment: production
# - Project: streaming-platform
# - Team: data-engineering
# - CostCenter: engineering

# Create enterprise budget alert
aws budgets create-budget --budget \
  --account-id 123456789 \
  --budget-name streaming-platform-monthly \
  --budget-limit Amount=30000,Unit=USD

# Alert if cost > $30K/month

// Order events at 1M/sec using AVRO schema
{
  "order_id": "ORD-1234567",
  "customer_id": "CUST-99999",
  "items": [...],
  "total_amount": 1299.99,
  "timestamp": "2025-10-26T10:00:00Z"
}

// Market data at 1M/sec
{
  "symbol": "AAPL",
  "price": 175.50,
  "volume": 10000,
  "exchange": "NASDAQ", 
  "timestamp": "2025-10-26T10:00:00.123Z"
}

// Sensor telemetry from millions of devices
// Using our optimized SensorData AVRO schema
{
  "sensorId": 1000001,
  "sensorType": 1,  // temperature sensor
  "temperature": 24.5,
  "humidity": 68.2,
  "pressure": 1013.25,
  "batteryLevel": 87.5,
  "status": 1,  // online
  "timestamp": 1635254400123
}

-- Real-time anomaly detection
CREATE MATERIALIZED VIEW anomaly_detection AS
SELECT 
    sensorId,
    AVG(temperature) as avg_temp,
    stddevPop(temperature) as stddev_temp,
    if(temperature > avg_temp + 3*stddev_temp, 1, 0) as is_anomaly
FROM benchmark.sensors_local
GROUP BY sensorId;

-- Enterprise windowed aggregations
CREATE MATERIALIZED VIEW hourly_metrics AS
SELECT 
    toStartOfHour(timestamp) as hour,
    sensorId,
    COUNT(*) as event_count,
    AVG(temperature) as avg_temp,
    MAX(temperature) as max_temp,
    MIN(temperature) as min_temp
FROM benchmark.sensors_local
GROUP BY hour, sensorId;

# Real-time ML inference with Flink
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.ml import Pipeline, KMeans

# Load trained model
model = Pipeline.load('s3://models/anomaly-detection')

# Apply to 1M events/sec stream
predictions = sensor_stream.map(lambda x: model.predict(x))

# Deploy to additional regions for global presence
# us-west-2 (primary)
# us-east-1 (DR)
# eu-west-1 (Europe)
# ap-southeast-1 (Asia)

# Enable Pulsar geo-replication
# Configure ClickHouse distributed tables
# Use Route53 for global load balancing