센서 데이터가 JSON으로 파싱되더라도 저장해도 되는 측정값이라는 뜻은 아닙니다. 0, null, 누락된 키와 문자열 "0"은 수신 과정에서 서로 다른 상태로 취급해야 합니다. 이 글에서는 가상 온도 센서의 데이터 계약을 JSON Schema로 작성하고 Python으로 경계값을 확인합니다.
확인일은 2026년 9월 14일입니다. 예제는 JSON Schema Draft 2020-12와 Python jsonschema 4.26.0으로 검증했습니다. 온도 범위와 상태 이름은 설명을 위해 정한 가상의 규칙입니다.
1. Data contract에 측정값의 의미를 담기
먼저 데이터 생산자와 소비자가 공유할 규칙을 정합니다. 아래 설계에서는 섭씨 단위를 필드명 temperature_c에 넣고, 측정 실패를 값과 별도의 quality로 표현합니다.
| 입력 상태 | 예제 계약의 처리 | 이유 |
|---|---|---|
| 숫자 0, quality=good | 수용 | 0도도 실제 측정값 |
| null, quality=missing | 수용 | 값이 없다는 상태를 명시 |
| temperature_c 키 누락 | 거부 | 메시지 구조 누락을 구별 |
| 문자열 "23.0" | 거부 | 송신 측 자료형 변경을 발견 |
| 숫자 23, quality=missing | 거부 | 값과 상태의 모순을 발견 |
필드를 properties에 선언하는 것과 필수로 만드는 것은 별개입니다. 필수 키는 required에 넣습니다. null이 들어 있는 키와 키 자체가 없는 경우도 다릅니다. 이 구분은 JSON Schema 객체 문서에 설명되어 있습니다.
이 계약을 적용하면 대시보드는 missing을 0도 그래프로 그리는 대신 데이터 공백으로 표시할 수 있습니다. 분석 파이프라인도 측정 실패 건수와 실제 영하·영상 온도를 따로 집계할 수 있습니다. 다만 실제 장비가 어떤 상태에서 null을 보내는지는 펌웨어와 수집 서버가 합의해야 합니다.
2. JSON Schema로 필수·범위·상태 연결하기
다음 내용을 UTF-8 파일 sensor.schema.json으로 저장합니다. -40부터 125까지는 이 예제의 가상 장비 범위이며 양 끝값을 포함합니다. 실제 제품에는 데이터시트와 측정 모드에 맞는 범위를 적용하십시오.
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"schema_version": {"const": 1},
"device_id": {"type": "string", "minLength": 1},
"temperature_c": {
"type": ["number", "null"],
"minimum": -40,
"maximum": 125
},
"quality": {"enum": ["good", "missing"]}
},
"required": ["schema_version", "device_id", "temperature_c", "quality"],
"additionalProperties": false,
"if": {"properties": {"quality": {"const": "good"}}},
"then": {"properties": {"temperature_c": {"type": "number"}}},
"else": {"properties": {"temperature_c": {"type": "null"}}}
}
숫자 또는 null을 허용한 다음, quality가 good이면 숫자만, missing이면 null만 허용하도록 좁힙니다. quality 자체를 필수 키와 enum으로 제한했으므로 다른 상태나 상태 누락은 통과하지 않습니다. additionalProperties: false는 오타로 생긴 필드도 거부합니다. 자료형·범위·상수 제약의 정의는 Draft 2020-12 검증 명세를 참고하십시오.
이 스키마가 측정값의 진실성까지 확인해 주지는 않습니다. 고장 난 센서가 범위 안의 숫자를 계속 보내면 형식 검사는 통과할 수 있습니다. 변화율, 센서 상태 레지스터, 교정 정보 같은 판단은 별도 계층에서 다뤄야 합니다.
3. Python에서 경계값과 실패 예제 실행하기
실습용 가상 환경에서 다음 순서로 실행합니다. 활성화 스크립트 대신 가상 환경의 Python 경로를 직접 사용합니다.
python -m venv .venv-schema
.\.venv-schema\Scripts\python.exe -m pip install "jsonschema==4.26.0"
.\.venv-schema\Scripts\python.exe .\validate_sensor.py
마지막 명령 전에 아래 파일을 validate_sensor.py로 저장하고 schema 파일과 같은 폴더에 둡니다. schema 파일은 스크립트 위치를 기준으로 읽습니다. check_schema는 스키마 자체의 형식을 먼저 확인하고 is_valid는 각 메시지를 검사합니다. API 사용법은 python-jsonschema 검증 문서에서 확인할 수 있습니다.
import json
from pathlib import Path
from jsonschema import Draft202012Validator
schema_path = Path(__file__).with_name("sensor.schema.json")
schema = json.loads(schema_path.read_text(encoding="utf-8"))
Draft202012Validator.check_schema(schema)
validator = Draft202012Validator(schema)
base = {
"schema_version": 1,
"device_id": "demo-01",
"temperature_c": 0,
"quality": "good",
}
cases = [
("zero", base, True),
("lower_bound", dict(base, temperature_c=-40), True),
("upper_bound", dict(base, temperature_c=125), True),
("too_high", dict(base, temperature_c=126), False),
("string_number", dict(base, temperature_c="23.0"), False),
("boolean", dict(base, temperature_c=True), False),
("missing_value", dict(base, temperature_c=None, quality="missing"), True),
("null_with_good", dict(base, temperature_c=None), False),
("number_with_missing", dict(base, quality="missing"), False),
("missing_key", {k: v for k, v in base.items() if k != "temperature_c"}, False),
("extra_key", dict(base, temprature_c=20), False),
("unknown_version", dict(base, schema_version=2), False),
]
for name, payload, expected in cases:
actual = validator.is_valid(payload)
if actual != expected:
raise RuntimeError(name + " produced an unexpected result")
print(name, "ACCEPT" if actual else "REJECT")
print("verified", len(cases), "cases")
zero·lower_bound·upper_bound·missing_value의 4개는 ACCEPT, 나머지 8개는 REJECT가 나와야 합니다. 마지막 줄은 verified 12 cases입니다. 이 결과는 위 가상 데이터의 로컬 검증 결과이며 실제 센서 성능이나 운영 수집 시스템의 검증 결과는 아닙니다.
4. Version 변경을 배포 순서에 연결하기
schema_version을 기록해도 수신기가 모든 버전을 이해하는 것은 아닙니다. 수신 단계에서 지원 버전을 조회하고 해당 스키마를 선택하는 과정이 필요합니다. 이 예제는 버전 1만 수용하므로 버전 2 메시지는 별도 보관하거나 명확한 오류로 처리할 수 있습니다.
현재처럼 추가 키를 금지하면 새 필드 하나를 더해도 기존 검증기가 거부합니다. 새 필드를 허용할 계획이라면 처음부터 확장 필드 공간을 정하거나, 새 스키마를 이해하는 수신기를 먼저 배포한 뒤 송신기를 전환하는 정책을 마련해야 합니다. 엄격한 검사가 호환성 비용을 만든다는 점을 배포 계획에 반영하십시오.
기존 temperature_c의 뜻을 섭씨에서 화씨로 바꾸는 변경은 필드 이름과 자료형이 같아도 위험합니다. 단위와 의미가 바뀌면 새 필드나 새 버전으로 구분하고 이전 버전의 원본 데이터를 보존하는 편이 과거 로그 재처리에 도움이 됩니다.
5. Ingestion에서 추가로 확인할 것
네트워크에서 들어온 문자열은 JSON 파싱 단계도 관리해야 합니다. Python 기본 JSON 파서는 NaN·Infinity 같은 비표준 상수를 허용하므로 엄격한 입력이 필요하면 parse_constant에서 거부하는 정책을 추가하십시오. 이 글의 테스트는 그런 상수를 포함하지 않는, 미리 만든 Python 객체를 대상으로 합니다. JSON 파싱 동작은 Python json 공식 문서에 설명되어 있습니다.
중복 키, 메시지 크기, 중첩 깊이 제한도 수신 계층의 별도 과제입니다. 실패 로그에는 장비 ID·스키마 버전·오류 경로를 남기되 인증 토큰이나 전체 민감 메시지를 그대로 기록하지 않는 구성을 권합니다. 날짜 문자열에 format을 추가할 때도 사용 검증기의 FormatChecker 설정을 확인해야 합니다.
파일에서 특정 값만 살펴보는 작업에는 jq 공식 도구가 편리합니다. 필터링과 계약 검증을 각각 실행하면 “JSON은 읽히는데 저장 단계에서 거부되는 이유”를 좁히기 쉽습니다. 실제 배포 전에는 위 경계값에 프로젝트의 단위 변경·알 수 없는 버전·누락 상태 사례를 보태 회귀 검증 데이터로 보관하십시오.