Real-Time Industrial Equipment Health Monitoring

Equipment failures in industrial facilities are rarely sudden — they announce themselves through gradual changes in vibration, temperature, current draw, and acoustic signatures. RisingWave, a PostgreSQL-compatible streaming database, continuously analyzes these signals in real time so maintenance teams can intervene before failures become production stops.

Why Real-Time Equipment Health Monitoring Matters

Unplanned downtime is one of the most costly events in manufacturing and process industries. When a critical pump, compressor, or conveyor fails without warning, the consequences extend far beyond the repair cost: production lines halt, downstream processes queue up, and in some industries, materials in process may be scrapped. Industry studies consistently place unplanned downtime costs at multiples of planned maintenance costs.

Traditional equipment monitoring relies on periodic manual inspections and fixed-threshold alarms on individual sensors. A vibration alarm fires when vibration exceeds a static limit — but a motor that is gradually trending toward that limit over two weeks is invisible until the alarm fires. By that point, bearing damage may already be severe and a replacement more expensive than intervention would have been a week earlier.

Health-index monitoring addresses this gap by computing composite indicators from multiple sensor streams and tracking how those indicators trend over time. A motor health score that degrades by 2% per day will cross a warning threshold well before it crosses an alarm threshold — giving maintenance teams a ten-day window to schedule intervention rather than reacting to a failed shutdown.

How Streaming SQL Enables Equipment Health Monitoring

RisingWave ingests multi-sensor telemetry from Kafka topics and maintains continuously updated health indicator views. Window functions compute rolling baseline statistics (mean and standard deviation over the last 24 hours) that serve as context for evaluating current readings. Temporal joins correlate sensor readings against equipment metadata — rated capacity, installation date, last service date — stored in a PostgreSQL asset management database.

Because health monitoring requires looking at trends across time (not just point-in-time values), RisingWave's HOP windows are especially useful: a HOP window with a 1-hour interval and 24-hour width produces an overlapping series of 24-hour summaries, updated every hour, that tracks how aggregate sensor statistics are evolving.

Building It Step by Step

Step 1: Create the Data Source

-- Multi-sensor equipment telemetry
CREATE SOURCE equipment_telemetry (
    equipment_id    VARCHAR,
    asset_tag       VARCHAR,
    facility_id     VARCHAR,
    equipment_type  VARCHAR,   -- MOTOR, PUMP, COMPRESSOR, CONVEYOR, HVAC
    sensor_type     VARCHAR,   -- VIBRATION_X, VIBRATION_Y, TEMP_BEARING, TEMP_AMBIENT, CURRENT_A, CURRENT_B, CURRENT_C, PRESSURE_IN, PRESSURE_OUT
    value           DOUBLE PRECISION,
    unit            VARCHAR,
    event_ts        TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'iot.equipment_telemetry',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

-- Equipment asset master from PostgreSQL (maintenance schedules, rated specs)
CREATE SOURCE equipment_assets (
    equipment_id        VARCHAR,
    asset_tag           VARCHAR,
    equipment_type      VARCHAR,
    rated_power_kw      DOUBLE PRECISION,
    installation_date   DATE,
    last_service_date   DATE,
    next_service_date   DATE,
    criticality_tier    SMALLINT  -- 1=Critical, 2=Important, 3=Standard
) WITH (
    connector = 'postgres-cdc',
    hostname = 'cmms-db',
    port = '5432',
    username = 'replicator',
    password = 'secret',
    database.name = 'cmms',
    schema.name = 'public',
    table.name = 'equipment_assets'
) FORMAT DEBEZIUM ENCODE JSON;

Step 2: Build the Core Materialized View

-- Latest reading per equipment-sensor combination
CREATE MATERIALIZED VIEW equipment_sensor_latest AS
SELECT DISTINCT ON (equipment_id, sensor_type)
    equipment_id,
    asset_tag,
    facility_id,
    equipment_type,
    sensor_type,
    value               AS latest_value,
    unit,
    event_ts            AS last_ts,
    NOW() - event_ts    AS data_age
FROM equipment_telemetry
ORDER BY equipment_id, sensor_type, event_ts DESC;

-- 1-hour rolling statistics per equipment-sensor (HOP: 1hr interval, 24hr width)
CREATE MATERIALIZED VIEW equipment_rolling_stats AS
SELECT
    window_start,
    window_end,
    equipment_id,
    sensor_type,
    AVG(value)      AS rolling_avg,
    STDDEV(value)   AS rolling_stddev,
    MIN(value)      AS rolling_min,
    MAX(value)      AS rolling_max,
    COUNT(*)        AS reading_count
FROM HOP(equipment_telemetry, event_ts, INTERVAL '1 HOUR', INTERVAL '24 HOURS')
GROUP BY window_start, window_end, equipment_id, sensor_type;

Step 3: Add Alerts and Aggregations

-- Composite health score per equipment (lower = worse health)
CREATE MATERIALIZED VIEW equipment_health_score AS
SELECT
    l.equipment_id,
    l.asset_tag,
    l.facility_id,
    l.equipment_type,
    a.criticality_tier,
    -- Normalize each sensor relative to 24h rolling stats
    AVG(
        CASE WHEN r.rolling_stddev > 0
             THEN GREATEST(0, 100 - 10 * ABS(l.latest_value - r.rolling_avg) / r.rolling_stddev)
             ELSE 100
        END
    ) AS health_score,
    COUNT(DISTINCT l.sensor_type) AS active_sensor_count,
    MAX(l.last_ts) AS last_reading_ts
FROM equipment_sensor_latest l
JOIN equipment_rolling_stats r
    ON  r.equipment_id  = l.equipment_id
    AND r.sensor_type   = l.sensor_type
    AND r.window_end    = (
        SELECT MAX(window_end) FROM equipment_rolling_stats
        WHERE equipment_id = l.equipment_id
          AND sensor_type  = l.sensor_type
    )
JOIN equipment_assets a ON a.equipment_id = l.equipment_id
GROUP BY l.equipment_id, l.asset_tag, l.facility_id, l.equipment_type, a.criticality_tier;

-- Health degradation alert: critical equipment below 80 or important below 70
CREATE MATERIALIZED VIEW health_degradation_alerts AS
SELECT
    equipment_id,
    asset_tag,
    facility_id,
    equipment_type,
    criticality_tier,
    health_score,
    active_sensor_count,
    last_reading_ts,
    CASE
        WHEN criticality_tier = 1 AND health_score < 80 THEN 'CRITICAL_ALERT'
        WHEN criticality_tier = 2 AND health_score < 70 THEN 'WARNING_ALERT'
        WHEN criticality_tier = 3 AND health_score < 60 THEN 'INFO_ALERT'
    END AS alert_level
FROM equipment_health_score
WHERE (criticality_tier = 1 AND health_score < 80)
   OR (criticality_tier = 2 AND health_score < 70)
   OR (criticality_tier = 3 AND health_score < 60);

-- Bearing temperature trend: rising trend over 2 consecutive hours
CREATE MATERIALIZED VIEW bearing_temp_trend AS
SELECT
    equipment_id,
    sensor_type,
    window_end                                       AS period_end,
    rolling_avg                                      AS avg_temp,
    LAG(rolling_avg) OVER (
        PARTITION BY equipment_id, sensor_type
        ORDER BY window_end
    )                                                AS prev_avg_temp,
    rolling_avg - LAG(rolling_avg) OVER (
        PARTITION BY equipment_id, sensor_type
        ORDER BY window_end
    )                                                AS temp_delta
FROM equipment_rolling_stats
WHERE sensor_type LIKE 'TEMP_%';

Step 4: Sink Results Downstream

-- Push health alerts to CMMS work order creation system
CREATE SINK health_alert_sink
FROM health_degradation_alerts
WITH (
    connector = 'kafka',
    topic = 'maintenance.alerts.health',
    properties.bootstrap.server = 'kafka:9092'
) FORMAT PLAIN ENCODE JSON;

-- Write health scores to dashboard database
CREATE SINK health_score_sink
FROM equipment_health_score
WITH (
    connector = 'jdbc',
    jdbc.url = 'jdbc:postgresql://dashboard-db:5432/maintenance',
    table.name = 'equipment_health_score'
) FORMAT PLAIN ENCODE JSON;

How This Compares to Traditional Approaches

FAQ

What is industrial equipment health monitoring?

Industrial equipment health monitoring is the continuous assessment of machine condition using multi-sensor telemetry data. It goes beyond simple threshold alarms to compute composite health indices, detect trends, and predict degradation before failures occur. The goal is to maximize equipment uptime while minimizing both reactive repairs and unnecessary preventive maintenance.

How does RisingWave compute rolling baselines efficiently?

RisingWave uses incremental computation for window functions. A HOP window with a 24-hour width does not recompute 24 hours of history on every new reading — it maintains an internal state that is updated with each new event and expired events are removed as windows slide. This keeps per-reading processing cost constant.

Can I integrate RisingWave with my existing stack?

Yes. RisingWave connects to Kafka, PostgreSQL via CDC, and MySQL via CDC. It writes results to JDBC-compatible databases and Kafka topics. CMMS platforms, ERP systems, and dashboarding tools that speak PostgreSQL can query health scores and alerts directly.

What happens when a sensor goes offline?

The data_age column in equipment_sensor_latest immediately reflects the sensor silence. You can build a separate view filtering on data_age > INTERVAL '5 MINUTES' to detect and alert on sensor connectivity loss.

Key Takeaways

• Fixed-threshold alarms fire too late; composite health scores computed from rolling baselines provide days of lead time before equipment failure.
• RisingWave's HOP windows maintain 24-hour rolling statistics incrementally — no full recomputation on each new reading.
• Temporal joins with the equipment asset master (from PostgreSQL CDC) enrich health scores with criticality tier, enabling tiered alerting thresholds.
• Health alerts flow to CMMS work order systems via Kafka sinks, automating the maintenance dispatch workflow end to end.

Ready to try this? Get started with RisingWave. Join our Slack community.