A dissolved oxygen (DO) crash below 5 mg/L can kill an entire fish pen within hours. RisingWave, a PostgreSQL-compatible streaming database, lets aquaculture operators ingest sensor telemetry from tanks and sea cages, compute multi-parameter water quality indices in real time, and trigger aeration or feed-cut alerts before mortality occurs.
Why Aquaculture Water Monitoring Matters
Water quality management is the single largest operational risk in finfish and shrimp farming. Key parameters drift continuously with weather, feeding cycles, and tidal exchange. Critical thresholds are narrow:
- Dissolved oxygen (DO): 6–9 mg/L is optimal for most salmonids; below 5 mg/L is dangerous
- Salinity: 28–35 ppt for marine species; stress begins outside a ±3 ppt swing
- pH: 7.5–8.5; ammonia toxicity increases sharply above pH 8.5
- Temperature: species-specific; a 2 °C deviation can trigger disease outbreaks
- Feed conversion ratio (FCR): tracked over days; rising FCR signals poor water quality or disease before mortality appears
Traditional water quality loggers record data every 15–30 minutes and require staff to review trend charts manually. A streaming pipeline can evaluate parameter combinations continuously and act within seconds of a dangerous drift.
How Streaming SQL Solves This
RisingWave ingests sensor events from Kafka, MQTT bridges, or HTTP sources and evaluates materialized views incrementally. Window functions compute rolling averages that smooth transient spikes, while multi-parameter composite indices capture interactions between parameters — for example, DO and temperature together determine saturation percentage, which is more informative than raw DO alone.
Because RisingWave is PostgreSQL-compatible, existing farm management software can query materialized views directly without API changes.
Step-by-Step Tutorial
Step 1: Data Source
Create a source for water quality sensor readings. Each sensor is installed in a specific pen and reports several parameters per reading.
CREATE SOURCE water_quality_readings (
sensor_id VARCHAR,
pen_id VARCHAR,
site_id VARCHAR,
do_mg_l DOUBLE PRECISION, -- dissolved oxygen mg/L
salinity_ppt DOUBLE PRECISION, -- salinity parts per thousand
temp_celsius DOUBLE PRECISION,
ph DOUBLE PRECISION,
turbidity_ntu DOUBLE PRECISION,
event_time TIMESTAMPTZ
)
WITH (
connector = 'kafka',
topic = 'aquaculture.water.quality',
properties.bootstrap.server = 'kafka:9092',
scan.startup.mode = 'latest'
)
FORMAT PLAIN ENCODE JSON
( TIMESTAMP COLUMN = event_time );
-- Feed conversion tracking (batch updates from feeding system)
CREATE SOURCE feed_events (
pen_id VARCHAR,
feed_kg DOUBLE PRECISION,
biomass_kg DOUBLE PRECISION, -- estimated pen biomass
event_time TIMESTAMPTZ
)
WITH (
connector = 'kafka',
topic = 'aquaculture.feed.events',
properties.bootstrap.server = 'kafka:9092',
scan.startup.mode = 'latest'
)
FORMAT PLAIN ENCODE JSON
( TIMESTAMP COLUMN = event_time );
Step 2: Core View
Compute 10-minute rolling averages per pen, detect parameter violations, and calculate a water quality index (WQI) score.
CREATE MATERIALIZED VIEW pen_water_averages AS
SELECT
pen_id,
site_id,
window_start,
window_end,
AVG(do_mg_l) AS avg_do,
MIN(do_mg_l) AS min_do,
AVG(salinity_ppt) AS avg_salinity,
AVG(temp_celsius) AS avg_temp,
AVG(ph) AS avg_ph,
AVG(turbidity_ntu) AS avg_turbidity,
COUNT(*) AS readings
FROM TUMBLE(water_quality_readings, event_time, INTERVAL '10 MINUTES')
GROUP BY pen_id, site_id, window_start, window_end;
CREATE MATERIALIZED VIEW water_quality_index AS
SELECT
pen_id,
site_id,
window_start,
window_end,
avg_do,
avg_salinity,
avg_temp,
avg_ph,
-- DO score: 0 = critical, 1 = optimal
CASE
WHEN avg_do < 5.0 THEN 0.0
WHEN avg_do < 6.0 THEN (avg_do - 5.0)
WHEN avg_do <= 9.0 THEN 1.0
WHEN avg_do < 11.0 THEN (11.0 - avg_do) / 2.0
ELSE 0.5
END AS do_score,
-- Salinity score
CASE
WHEN avg_salinity < 25.0 OR avg_salinity > 38.0 THEN 0.0
WHEN avg_salinity BETWEEN 28.0 AND 35.0 THEN 1.0
ELSE 0.5
END AS salinity_score,
-- pH score
CASE
WHEN avg_ph < 7.0 OR avg_ph > 9.0 THEN 0.0
WHEN avg_ph BETWEEN 7.5 AND 8.5 THEN 1.0
ELSE 0.5
END AS ph_score,
-- Composite WQI (weighted)
(
CASE WHEN avg_do < 5.0 THEN 0.0 WHEN avg_do < 6.0 THEN (avg_do-5.0) WHEN avg_do<=9.0 THEN 1.0 WHEN avg_do<11.0 THEN (11.0-avg_do)/2.0 ELSE 0.5 END * 0.5 +
CASE WHEN avg_salinity < 25.0 OR avg_salinity > 38.0 THEN 0.0 WHEN avg_salinity BETWEEN 28.0 AND 35.0 THEN 1.0 ELSE 0.5 END * 0.25 +
CASE WHEN avg_ph < 7.0 OR avg_ph > 9.0 THEN 0.0 WHEN avg_ph BETWEEN 7.5 AND 8.5 THEN 1.0 ELSE 0.5 END * 0.25
) AS wqi
FROM pen_water_averages;
-- Rolling feed conversion ratio per pen (7-day window)
CREATE MATERIALIZED VIEW pen_fcr AS
SELECT
pen_id,
window_start,
window_end,
SUM(feed_kg) AS total_feed_kg,
AVG(biomass_kg) AS avg_biomass_kg,
SUM(feed_kg) / NULLIF(MAX(biomass_kg) - MIN(biomass_kg), 0) AS fcr
FROM TUMBLE(feed_events, event_time, INTERVAL '7 DAYS')
GROUP BY pen_id, window_start, window_end;
Step 3: Alerts and Sinks
Emit critical water quality alerts and route to the farm operations platform.
CREATE MATERIALIZED VIEW water_alerts AS
SELECT
pen_id,
site_id,
window_end AS alert_time,
avg_do,
avg_salinity,
avg_temp,
avg_ph,
wqi,
CASE
WHEN avg_do < 5.0 THEN 'CRITICAL_DO'
WHEN avg_do < 6.0 THEN 'LOW_DO'
WHEN avg_ph > 8.5 OR avg_ph < 7.0 THEN 'PH_OUT_OF_RANGE'
WHEN avg_salinity < 25.0 OR avg_salinity > 38.0 THEN 'SALINITY_DEVIATION'
WHEN wqi < 0.5 THEN 'POOR_WATER_QUALITY'
ELSE 'WARNING'
END AS alert_type,
CASE
WHEN avg_do < 5.0 THEN 'CRITICAL'
WHEN wqi < 0.5 THEN 'WARNING'
ELSE 'INFO'
END AS severity
FROM water_quality_index
WHERE wqi < 0.7 OR avg_do < 6.0;
CREATE SINK water_alerts_sink
FROM water_alerts
WITH (
connector = 'kafka',
topic = 'aquaculture.alerts',
properties.bootstrap.server = 'kafka:9092'
)
FORMAT PLAIN ENCODE JSON;
Comparison Table
| Approach | Latency | Multi-Parameter Index | FCR Tracking | SQL Interface |
| Manual log review | Hours | No | Manual | No |
| Data logger + daily export | 15–30 min | No | Spreadsheet | No |
| Custom stream processor | Seconds | Complex | Custom code | No |
| RisingWave streaming SQL | Seconds | Yes | Built-in | Yes |
FAQ
Q: Can RisingWave handle sensors at different sampling rates? Yes. Sensors reporting every 5 seconds and sensors reporting every 5 minutes can both feed the same source table. Window aggregations tolerate irregular arrival times, and watermark settings control how long to wait for late data.
Q: How do I incorporate tidal correction for salinity readings in sea cages? Create a reference table with tidal cycle data per site and join it to the water quality view. Subtract the expected tidal salinity component from the measured value before computing the salinity score.
Q: What is a realistic feed conversion ratio to flag? An FCR above 1.8 for Atlantic salmon or above 1.5 for shrimp typically indicates overfeeding, poor water quality, or disease. Store species-specific FCR thresholds in a lookup table and join them into the alert logic.
Key Takeaways
- RisingWave processes multi-parameter water quality telemetry with standard SQL — no Flink or Spark required.
- A composite Water Quality Index combining DO, salinity, and pH provides earlier and more reliable warnings than single-parameter thresholds.
- Feed conversion ratio tracking over 7-day windows adds a lagging biological signal that confirms water quality trends.
- Alerts route automatically to farm operations platforms via Kafka, enabling aeration and feeding interventions in time to prevent fish losses.
