Data Scientist Journey: Predictive Maintenance for CNC Machines

{
  "source": {
    "type": "kafka",
    "config": {
      "bootstrap_servers": "kafka.apex-iot.internal:9092",
      "topic_pattern": "sensors.cnc.*",
      "consumer_group": "matih-ingestion-sensors",
      "format": "avro",
      "schema_registry_url": "http://schema-registry:8081"
    }
  },
  "destination": {
    "type": "s3_parquet",
    "config": {
      "bucket": "apex-data-lake",
      "prefix": "raw/sensors/",
      "partition_by": ["plant_id", "date"],
      "compression": "snappy"
    }
  },
  "transform": {
    "aggregation_window": "5m",
    "aggregations": ["mean", "stddev", "min", "max", "count"],
    "group_by": ["machine_id", "sensor_type"]
  }
}

File: cnc_machine_specs.csv (5,247 rows)
Columns: machine_id, manufacturer, model, spindle_max_rpm, max_feed_rate,
         axis_count, coolant_type, rated_power_kw, weight_kg, install_date

Catalog: apex_manufacturing
  └── sensors/
      ├── sensor_readings_5min     (100M rows, 47 sensor types)
      │   ├── vibration_x          (g-force, range: 0.01 - 8.4)
      │   ├── vibration_y          (g-force, range: 0.01 - 7.9)
      │   ├── vibration_z          (g-force, range: 0.01 - 9.1)
      │   ├── spindle_temperature  (celsius, range: 22 - 94)
      │   ├── coolant_temperature  (celsius, range: 15 - 38)
      │   ├── coolant_flow_rate    (L/min, range: 0 - 45)
      │   ├── spindle_speed        (rpm, range: 0 - 12000)
      │   ├── feed_rate            (mm/min, range: 0 - 8000)
      │   ├── spindle_load         (%, range: 0 - 100)
      │   ├── power_consumption    (kW, range: 0 - 75)
      │   └── ... (37 more sensor types)
      └── sensor_readings_raw      (archived, not for direct query)

  MQTT Broker          Kafka Topic         5-min Aggregation        S3 Parquet
  ┌──────────┐        ┌──────────┐        ┌──────────────────┐     ┌──────────┐
  │ Raw      │───────▶│ sensors. │───────▶│ Pipeline:        │────▶│ sensors/ │
  │ signal   │        │ cnc.     │        │ sensor_agg_5min  │     │ 5min_agg │
  │ (100 Hz) │        │ plant_1  │        │                  │     │ .parquet │
  │          │        │          │        │ mean, stddev,    │     │          │
  │          │        │          │        │ min, max, count  │     │          │
  └──────────┘        └──────────┘        └──────────────────┘     └──────────┘

-- Rolling statistics over multiple time horizons
-- Uses DuckDB for efficient window functions on S3 parquet
SELECT
    machine_id,
    timestamp,
    -- 1-hour rolling statistics
    AVG(vibration_rms) OVER w1h AS vibration_mean_1h,
    STDDEV(vibration_rms) OVER w1h AS vibration_std_1h,
    MAX(vibration_rms) OVER w1h AS vibration_max_1h,
    -- 6-hour rolling statistics
    AVG(vibration_rms) OVER w6h AS vibration_mean_6h,
    STDDEV(vibration_rms) OVER w6h AS vibration_std_6h,
    -- 24-hour rolling statistics
    AVG(vibration_rms) OVER w24h AS vibration_mean_24h,
    STDDEV(vibration_rms) OVER w24h AS vibration_std_24h,
    -- Rate of change (trend detection)
    (AVG(vibration_rms) OVER w1h - AVG(vibration_rms) OVER w24h)
        / NULLIF(AVG(vibration_rms) OVER w24h, 0) AS vibration_trend_ratio,
    -- Temperature rate of change (degrees per hour)
    (MAX(spindle_temperature) OVER w1h - MIN(spindle_temperature) OVER w1h)
        AS temp_range_1h,
    -- Operating context
    AVG(spindle_speed) OVER w1h AS avg_spindle_speed_1h,
    AVG(spindle_load) OVER w1h AS avg_spindle_load_1h,
    -- Hours since last maintenance
    EXTRACT(EPOCH FROM (timestamp - last_maintenance_time)) / 3600.0
        AS hours_since_maintenance
FROM sensor_readings_5min s
LEFT JOIN (
    SELECT machine_id,
           MAX(end_time) AS last_maintenance_time
    FROM maintenance_logs
    WHERE maintenance_type IN ('PREVENTIVE', 'CORRECTIVE')
    GROUP BY machine_id
) m ON s.machine_id = m.machine_id
WINDOW
    w1h  AS (PARTITION BY machine_id ORDER BY timestamp
             RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW),
    w6h  AS (PARTITION BY machine_id ORDER BY timestamp
             RANGE BETWEEN INTERVAL '6' HOUR PRECEDING AND CURRENT ROW),
    w24h AS (PARTITION BY machine_id ORDER BY timestamp
             RANGE BETWEEN INTERVAL '24' HOUR PRECEDING AND CURRENT ROW)

-- Create target variable: time-to-failure for each sensor reading
-- A reading is labeled with the hours remaining until the next unplanned failure
WITH failure_events AS (
    SELECT
        machine_id,
        start_time AS failure_time,
        description AS failure_mode,
        CASE
            WHEN description ILIKE '%bearing%' THEN 'BEARING_WEAR'
            WHEN description ILIKE '%spindle%alignment%' THEN 'SPINDLE_ALIGNMENT'
            WHEN description ILIKE '%coolant%' THEN 'COOLANT_SYSTEM'
            WHEN description ILIKE '%electrical%' OR description ILIKE '%drive%'
                THEN 'ELECTRICAL'
            ELSE 'OTHER'
        END AS failure_category
    FROM maintenance_logs
    WHERE maintenance_type = 'CORRECTIVE'
      AND unplanned = true
)
SELECT
    s.*,
    f.failure_time,
    f.failure_category,
    EXTRACT(EPOCH FROM (f.failure_time - s.timestamp)) / 3600.0
        AS hours_to_failure
FROM sensor_features s
LEFT JOIN LATERAL (
    SELECT failure_time, failure_category
    FROM failure_events fe
    WHERE fe.machine_id = s.machine_id
      AND fe.failure_time > s.timestamp
    ORDER BY fe.failure_time ASC
    LIMIT 1
) f ON true

                    Apex Predictive Maintenance Pipeline

  STREAMING (continuous)                 BATCH (nightly at 2 AM)
  ┌─────────────────────────┐           ┌─────────────────────────┐
  │ Kafka sensor topic      │           │ S3 sensor parquet       │
  │         │               │           │         │               │
  │         ▼               │           │         ▼               │
  │ 5-min aggregation       │           │ Feature computation     │
  │         │               │           │ (rolling 1h/6h/24h)     │
  │         ▼               │           │         │               │
  │ Anomaly detection       │           │         ▼               │
  │ (Z-score > 3.5)         │           │ Model retraining check  │
  │         │               │           │ (drift detected?)       │
  │         ▼               │           │         │               │
  │ Real-time health score  │           │    ┌────┴─────┐         │
  │ (Ray Serve inference)   │           │    │ Yes      │ No      │
  │         │               │           │    ▼          ▼         │
  │         ▼               │           │ Retrain    Log metrics  │
  │ Alert if score < 0.3    │           │ model      and exit     │
  └─────────────────────────┘           └─────────────────────────┘

{
  "workflow": "predictive_maintenance_batch",
  "schedule": "0 2 * * *",
  "activities": [
    {
      "name": "compute_features",
      "type": "sql_transform",
      "config": {
        "query_file": "features/rolling_sensor_features.sql",
        "output_table": "ml.sensor_features_daily",
        "partition_by": "date"
      },
      "timeout": "45m"
    },
    {
      "name": "validate_features",
      "type": "data_quality",
      "config": {
        "suite": "sensor_feature_quality",
        "expectations": [
          {"type": "expect_column_values_to_not_be_null", "column": "vibration_mean_1h"},
          {"type": "expect_column_values_to_be_between", "column": "vibration_std_1h", "min": 0, "max": 10},
          {"type": "expect_table_row_count_to_be_between", "min": 50000}
        ],
        "fail_pipeline_on_violation": true
      }
    },
    {
      "name": "check_model_drift",
      "type": "model_monitoring",
      "config": {
        "model_name": "cnc_predictive_maintenance",
        "metrics": ["psi", "csi"],
        "threshold": 0.15,
        "reference_window": "30d"
      }
    },
    {
      "name": "retrain_if_drifted",
      "type": "conditional",
      "condition": "check_model_drift.drift_detected == true",
      "activity": {
        "name": "retrain_model",
        "type": "ml_training",
        "config": {
          "experiment": "cnc_predictive_maintenance",
          "training_data": "ml.sensor_features_daily",
          "model_type": "random_survival_forest",
          "hyperparameters": {
            "n_estimators": 500,
            "max_depth": 12,
            "min_samples_leaf": 20
          }
        }
      }
    }
  ]
}

  Failure Mode Distribution (Apex Manufacturing, 2 years)
  ════════════════════════════════════════════════════════

  BEARING_WEAR        ███████████████████████████████████████████  43%
  SPINDLE_ALIGNMENT   ████████████████████████████               28%
  COOLANT_SYSTEM      ██████████████████                         18%
  ELECTRICAL          ███████████                                11%

  Total failures: 847 events across 200 machines
  Failure rate: 2.3% of operating 5-min windows
  Mean time to failure after first anomaly: 72 hours (bearing), 48h (spindle)

-- Top sensor features correlated with impending bearing failure
-- Run in ML Workbench notebook
SELECT
    sensor_feature,
    CORR(feature_value, CASE WHEN hours_to_failure < 72 THEN 1 ELSE 0 END)
        AS correlation_with_failure,
    AVG(CASE WHEN hours_to_failure < 72 THEN feature_value END)
        AS avg_pre_failure,
    AVG(CASE WHEN hours_to_failure > 168 THEN feature_value END)
        AS avg_normal
FROM sensor_features_pivoted
WHERE failure_category = 'BEARING_WEAR'
GROUP BY sensor_feature
ORDER BY ABS(correlation_with_failure) DESC
LIMIT 10

{
  "experiment_name": "cnc_predictive_maintenance",
  "model_type": "random_survival_forest",
  "training_config": {
    "features": [
      "vibration_mean_1h", "vibration_std_1h", "vibration_max_1h",
      "vibration_mean_6h", "vibration_std_6h",
      "vibration_mean_24h", "vibration_std_24h", "vibration_trend_ratio",
      "spindle_temperature_mean_1h", "temp_range_1h",
      "coolant_flow_mean_1h", "spindle_load_std_6h",
      "power_consumption_mean_6h", "hours_since_maintenance",
      "machine_type", "machine_age_years"
    ],
    "target": "hours_to_failure",
    "event_column": "failure_occurred",
    "stratify_by": "failure_category",
    "train_test_split": 0.8,
    "cross_validation_folds": 5
  },
  "hyperparameters": {
    "n_estimators": 500,
    "max_depth": 12,
    "min_samples_leaf": 20,
    "max_features": "sqrt",
    "n_jobs": -1
  },
  "compute": {
    "type": "ray_train",
    "num_workers": 4,
    "resources_per_worker": {"cpu": 4, "memory_gb": 16}
  }
}

{
  "deployment": {
    "model_name": "cnc_predictive_maintenance",
    "model_version": "v3",
    "serving_framework": "ray_serve",
    "endpoint": "/api/v1/ml/predict/machine-health",
    "config": {
      "num_replicas": 2,
      "max_concurrent_queries": 100,
      "ray_actor_options": {
        "num_cpus": 2,
        "memory": 4294967296
      }
    },
    "input_schema": {
      "machine_id": "string",
      "sensor_features": "object (16 feature values)"
    },
    "output_schema": {
      "machine_id": "string",
      "health_score": "float (0-1, lower = higher risk)",
      "predicted_rul_hours": "float",
      "failure_probability_24h": "float",
      "failure_probability_72h": "float",
      "top_risk_factors": "array of {feature, contribution}"
    }
  }
}

{
  "machine_id": "M-0073",
  "health_score": 0.28,
  "predicted_rul_hours": 34.5,
  "failure_probability_24h": 0.12,
  "failure_probability_72h": 0.67,
  "failure_mode_probabilities": {
    "BEARING_WEAR": 0.81,
    "SPINDLE_ALIGNMENT": 0.11,
    "COOLANT_SYSTEM": 0.05,
    "ELECTRICAL": 0.03
  },
  "top_risk_factors": [
    {"feature": "vibration_std_24h", "contribution": 0.34, "value": 2.1, "normal_range": "0.2-0.6"},
    {"feature": "vibration_trend_ratio", "contribution": 0.22, "value": 0.28, "normal_range": "-0.05-0.05"},
    {"feature": "temp_range_1h", "contribution": 0.15, "value": 11.3, "normal_range": "2-5"}
  ],
  "recommended_action": "Schedule bearing inspection within 24 hours",
  "timestamp": "2026-02-28T14:30:00Z"
}

  Sensor Data ──▶ Ray Serve ──▶ Health Score < 0.3? ──▶ CMMS Work Order
       │              │               │                       │
       │              │          ┌────┴─────┐                 │
       │              │          │ Yes      │ No              │
       │              │          ▼          ▼                 │
       │              │    Generate WO   Continue             │
       │              │    with priority  monitoring           │
       │              │    and failure                         │
       │              │    mode context                        │

{
  "alerts": [
    {
      "name": "model_accuracy_degradation",
      "metric": "precision_at_recall_80",
      "condition": "< 0.70 for 3 consecutive evaluations",
      "action": "trigger_retraining_pipeline",
      "notify": ["lin.wei@apex.com", "#ml-ops-alerts"]
    },
    {
      "name": "false_positive_spike",
      "metric": "false_positive_rate",
      "condition": "> 15% daily",
      "action": "notify_and_log",
      "notify": ["lin.wei@apex.com", "maintenance-supervisor@apex.com"]
    },
    {
      "name": "feature_drift_detected",
      "metric": "population_stability_index",
      "condition": "> 0.15 on any feature",
      "action": "trigger_drift_investigation_workflow",
      "notify": ["lin.wei@apex.com"]
    }
  ]
}

-- Evaluate transfer learning: train on lathes, test on mills
SELECT
    e.machine_type AS test_machine_type,
    COUNT(*) AS test_samples,
    AVG(CASE WHEN predicted_failure AND actual_failure THEN 1.0 ELSE 0.0 END)
        AS precision,
    AVG(CASE WHEN predicted_failure AND actual_failure THEN 1.0
         WHEN NOT predicted_failure AND actual_failure THEN 0.0
         ELSE NULL END) AS recall
FROM model_predictions p
JOIN equipment_registry e ON p.machine_id = e.machine_id
WHERE p.experiment_id = 'transfer_learning_lathe_to_mill'
GROUP BY e.machine_type