Smart Agriculture IoT Platform

The core challenge was handling 50M+ daily readings while maintaining real-time responsiveness:

Kafka Topic Design:

• sensor-raw — Unvalidated sensor readings (high volume)
• sensor-validated — Passed validation, enriched with metadata
• sensor-alerts — Threshold violations and anomalies
• sensor-recommendations — Actionable insights for facilities

Stream Processing Stages:

1. Validation — Schema validation, range checks, sensor health
2. Enrichment — Add plant type, zone info, historical context
3. Aggregation — 5-minute, hourly, and daily rollups
4. Analysis — Trend detection, anomaly scoring, predictions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

@app.agent(sensor_raw_topic)
async def process_reading(readings):
    async for reading in readings:
        # Validate
        if not validate_reading(reading):
            await sensor_invalid.send(value=reading)
            continue
        
        # Enrich with plant/zone metadata
        enriched = await enrich_reading(reading)
        
        # Check for anomalies
        anomaly_score = await anomaly_detector.score(enriched)
        enriched.anomaly_score = anomaly_score
        
        await sensor_validated.send(value=enriched)

Moving from PostgreSQL to TimescaleDB required careful planning:

Hypertable Design:

• Partitioned by time (1 week chunks) and sensor_id
• Compression enabled for data older than 7 days
• Retention policy: raw data 90 days, aggregates 2 years

Continuous Aggregates:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

CREATE MATERIALIZED VIEW sensor_hourly
WITH (timescaledb.continuous) AS
SELECT 
  time_bucket('1 hour', timestamp) AS bucket,
  sensor_id,
  AVG(moisture) as avg_moisture,
  MIN(moisture) as min_moisture,
  MAX(moisture) as max_moisture,
  COUNT(*) as reading_count
FROM sensor_readings
GROUP BY bucket, sensor_id;

Query Performance:

• 7-day range query: 45s → 120ms
• 30-day aggregation: 3min → 800ms
• Dashboard load: 5s → 200ms