Schema Validator

The schema is defined in a JSON file. Here is the full structure:

{
    "allowed_measurements": ["weather", "cpu"],

    "tables": {
        "weather": {
            "target_table": "weather_clean",
            "tags": {
                "location": {
                    "required": true,
                    "allowed_values": ["us-east", "us-west", "eu-west"]
                },
                "station_id": {
                    "required": true
                },
                "region": {
                    "required": false
                }
            },
            "fields": {
                "temperature": {
                    "required": true,
                    "type": "float"
                },
                "humidity": {
                    "required": true,
                    "type": "float"
                },
                "condition": {
                    "required": false,
                    "type": "string",
                    "allowed_values": ["sunny", "cloudy", "rain", "snow"]
                }
            }
        }
    }
}

IoT Sensor Validation

Ensure sensor readings always have a device_id, valid sensor type, and a numeric value:

{
    "allowed_measurements": ["sensor_readings"],
    "tables": {
        "sensor_readings": {
            "target_table": "sensors_validated",
            "tags": {
                "device_id": { "required": true },
                "sensor_type": {
                    "required": true,
                    "allowed_values": ["temperature", "pressure", "humidity"]
                }
            },
            "fields": {
                "value": { "required": true, "type": "float" },
                "status": {
                    "required": false,
                    "type": "string",
                    "allowed_values": ["ok", "warning", "critical"]
                }
            }
        }
    }
}

Multi-table with Cross-database

Validate weather and cpu data from raw_db, writing clean data to clean_db:

influxdb3 create trigger \
  --database raw_db \
  --plugin-filename schema_validator.py \
  --trigger-spec \"all_tables\" \
  --trigger-arguments schema_file=schema_validator_config.json,target_database=clean_db,write_rejection_log=true \
  schema_validator_all

Monitoring Rejections

Query the rejection log to see what data is being rejected and why:

SELECT * FROM _schema_rejections
WHERE time > now() - INTERVAL '1 hour'
ORDER BY time DESC