IoT Data Ingestion at Any Scale

-- Create a pipeline for sensor data with 1-minute aggregations
CREATE PIPELINE sensor_metrics AS
SELECT
  device_id,
  sensor_type,
  window_start,
  window_end,
  AVG(value) as avg_value,
  MIN(value) as min_value,
  MAX(value) as max_value,
  COUNT(*) as reading_count
FROM TABLE(
  TUMBLE(
    TABLE sensor_events,
    DESCRIPTOR(event_time),
    INTERVAL '1' MINUTE
  )
)
GROUP BY device_id, sensor_type, window_start, window_end
INTO iceberg.sensors.metrics_1min;

-- Detect anomalies in real-time
CREATE PIPELINE anomaly_alerts AS
SELECT
  device_id,
  sensor_type,
  value,
  event_time,
  'HIGH_TEMP_ALERT' as alert_type
FROM sensor_events
WHERE sensor_type = 'temperature' AND value > 100
INTO iceberg.sensors.alerts;