지식그래프 입문 (2편) — WAMIS 공개 API와 Neo4j로 대전 하천 지식그래프 구축하기

About this offer

들어가며 1편 에서 지식그래프와 온톨로지의 차이, 구축 절차를 정리했다. 이번 글은 그 절차대로 대전의 실제 하천·수위관측소 데이터로 지식그래프를 만들어 본 기록이다. 처음에는 가상의 도시를 하나 만들어서 가짜 하천·관측소 데이터로 연습했다. 그걸로는 잘 돌아갔는데, 실제 데이터로 바꾸자마자 문제가 줄줄이 나왔다. 원본에 값이 빠져 있기도 하고, 출처마다 값이 달랐고, 관측소마다 수위 기준이 다른 경우도 있었다. 구축 과정과 함께 이런 문제를 어떻게 처리했는지도 같이 적었다. 작업 환경은 Windows 11이고, Docker(Rancher Desktop)로 Neo4j 5.26 Community를 띄웠다. 데이터 수집은 Python 3.11 표준 라이브러리만으로 했다. 1) 이번 글에서 다룰 내용 답할 질문 정하기와 그래프 모델 설계 (노드 3종, 관계 4종) WAMIS API로 대전 관측소 19곳 수집 ( fetch_wamis.py ) Docker로 Neo4j 띄우고 LOAD CSV + MERGE 로 적재 가변 길이 경로 질의로 "금강까지 몇 단계?" 같은 질문에 답하기 트러블슈팅 — 경로 변환, BOM, 콘솔 인코딩, 원천 결측, 집계 중복, 수위 기준 2) 배경 / 설계 2-1. 답할 질문 1편에서 정리한 대로 질문부터 적었다. 하천이 어디로 흘러가는지, 넘치면 어디를 봐야 하는지처럼 연결을 따라가야 답이 나오는 질문 위주로 골랐다. # 질문 필요한 노드·관계 Q1 금강으로 흘러드는 대전 하천은 모두 무엇인가? (몇 단계든) 하천, 지류이다 Q2 유등천이 넘치면 어느 구, 어느 관측소를 봐야 하나? (상류 포함) 하천, 관측소, 행정구역 Q3 하천마다 합류 직전의 관측소는 어디인가? 관측소의 합류점거리 Q4 구별로 관측소가 몇 곳이고 어떤 하천을 보나? 관측소, 위치한다 Q5 어떤 하천과도 연결되지 않은 관측소가 있나? (품질 점검) 관측한다 2-2. 데이터 출처 관측소 정보는 처음에 웹 검색으로 모으려고 했다. 검색 결과에 관측소 이름이랑 주의보 수위 같은 값이 요약돼서 나오길래 금방 될 줄 알았는데, 근거 페이지를 열어 보니 404였다. 원문으로 확인이 안 되는 숫자를 실제 데이터라고 넣을 수는 없어서 API로 방향을 바꿨다. API도 처음엔 HRFCO 시리즈 때 쓰던 홍수통제소 API를 다시 쓰려고 했다. 그런데 인증키를 다시 꺼내서 설정하는 게 귀찮아서 키 없이 되는 게 있나 찾아봤고, WAMIS(국가수자원관리종합정보시스템)에서 관측소 제원이랑 시자료를 키 없이 준다는 걸 알게 됐다. 게다가 제원에 하천명까지 들어 있어서 오히려 이쪽이 더 편했다. 데이터 출처 비고 관측소 19곳 제원·최신 수위 WAMIS Open API ( wl_dubwlobs , wl_obsinfo , wl_hrdata ) 인증키 불필요, 스크립트로 수집 국가하천 4개 연장·합류점 대전시 하천관리사업소 하천현황 금강·갑천·유등천·대전천 갑천 지천 대전시 3대하천현황 진잠천·유성천·관평천·매노천 등 대동천→대전천, 반석천→유성천 위키백과 대전시 자료에 없는 2건 보완 출처를 확인 못 한 값은 넣지 않고 비워 두기로 했다. 이 원칙 때문에 뒤에서 몇 번 손해를 보긴 했는데, 그래도 틀린 값이 섞이는 것보다는 낫다고 봤다. 2-3. 그래프 모델 (관측소)-[:관측한다]->(하천) (관측소)-[:위치한다]->(행정구역) (하천)-[:흐른다]->(행정구역) ← 관측소 주소(양쪽 강변)에서 도출 (하천)-[:지류이다]->(하천) ← 몇 단계든 따라갈 수 있는 고리 라벨과 관계 타입은 한글로 지었다. Cypher가 유니코드 식별자를 허용해서 백틱 없이 (:하천) , [:지류이다] 로 쓸 수 있다. 업무 담당자랑 같이 질의를 볼 일이 있으면 한글 쪽이 훨씬 읽기 편하다. 2-4. 전체 흐름과 프로젝트 구조 knowledge-graph-lab/ ├── scripts/fetch_wamis.py # WAMIS → stations.csv, station_regions.csv ├── data/ │ ├── stations.csv # 관측소 19 (스크립트 생성) │ ├── station_regions.csv # 관측소-구 26 (스크립트 생성) │ ├── rivers.csv # 하천 10 (출처 컬럼 포함) │ ├── river_tributary.csv # 합류 9 (출처 컬럼 포함) │ └── regions.csv # 자치구 5 ├── neo4j/docker-compose.yml ├── cypher/ │ ├── 01_schema.cypher # 유일성 제약조건 │ ├── 02_load.cypher # CSV → 노드·관계 │ └── 03_queries.cypher # Q1~Q5 └── SOURCES.md # 항목별 출처와 알려진 한계 3) 목표 상태 항목 값 노드 34 (관측소 19 · 하천 10 · 행정구역 5) 관계 60 (관측한다 18 · 위치한다 19 · 지류이다 9 · 흐른다 14) 검증 원천 CSV 행 수와 노드·관계 수 일치 질의 Q1~Q5 결과가 원천 자료로 손 검산한 값과 일치 4) 데이터 분석 — WAMIS 응답 살펴보기 4-1. 관측소 목록 → 대전만 고르기 wl_dubwlobs 를 호출하면 전국 관측소 목록 1,363건이 온다. 관측소명이 대전시(원촌교) 형식이라 대전시( 로 시작하는 것만 골랐더니 19곳이었다. http://www.wamis.go.kr:8080/wamis/openapi/wkw/wl_dubwlobs?output=json 4-2. 관측소 제원 — 하천명이 들어 있다 wl_obsinfo 에 관측소 코드를 넣으면 제원이 나온다. 여기서 제일 반가웠던 게 rivnm (하천명) 필드다. 관측소 → 하천 관계를 다른 자료 없이 바로 만들 수 있었다. 그래프에 쓴 필드는 이 정도다. 필드 예 (원촌교) 처리 wlobscd 3009680 관측소 노드 식별자 rivnm 갑천 관측한다 관계 addr 대전광역시 대덕구 대화동 원촌교 (대전광역시시 유성구 원촌동) 괄호 밖 = 본주소 구, 괄호 안 = 맞은편 강변 구 lat / lon 36-22-42 / 127-24-36 도-분-초 문자열 → 십진수, point() 로 저장 bsnara 609.4 유역면적(km2) rvmjctdis 10.32 합류점까지 거리(km) — Q3에 사용 주소를 보다가 괄호 안에 다리 건너편 주소가 들어 있다는 걸 알았다. 하천은 구 경계를 따라 흐르는 경우가 많아서, 괄호 안 구까지 모으면 "이 하천이 지나는 구"를 어느 정도 만들 수 있다. 원본에는 대전광역시시 같은 오타나 lat 끝에 붙은 공백도 있어서, 공백 기준으로 잘라서 구 이름만 고르고 strip() 을 했다. 4-3. 시자료 수위 wl_hrdata 는 관측소별 시간 단위 수위를 준다. 최근 이틀 치를 받아서 값이 있는 마지막 행을 최신 수위로 썼다. .../wl_hrdata?obscd=3009680&startdt=20261006&enddt=20261007&output=json → [{"ymdh": "2026100601", "wl": ".45"}, ..., {"ymdh": "2026100709", "wl": ".40"}] 수위가 ".45" 처럼 앞의 0이 빠진 문자열로 와서 float() 로 바꿨다. 5) 해결 (구현) 5-1. 수집 스크립트 — fetch_wamis.py 표준 라이브러리만 썼다. 관측소 CSV와 관측소-구 CSV 두 개를 만든다. """WAMIS(국가수자원관리종합정보시스템) Open API 에서 대전 수위관측소 데이터를 받아 CSV 로 만든다. 인증키가 필요 없다. 실행: python scripts/fetch_wamis.py 결과: data/stations.csv, data/station_regions.csv """ import csv import json import re import urllib.request from datetime import datetime, timedelta from pathlib import Path BASE = "http://www.wamis.go.kr:8080/wamis/openapi/wkw" DATA = Path(__file__).resolve().parent.parent / "data" DISTRICTS = ["동구", "중구", "서구", "유성구", "대덕구"] def get(endpoint, **params): query = "&".join(f"{k}={v}" for k, v in {**params, "output": "json"}.items()) with urllib.request.urlopen(f"{BASE}/{endpoint}?{query}", timeout=30) as res: return json.load(res) def dms_to_decimal(dms): """'127-24-36' → 127.41""" d, m, s = (float(x) for x in dms.strip().split("-")) return round(d + m / 60 + s / 3600, 6) def districts_in(addr): """주소에서 대전 자치구를 순서대로 뽑는다. 괄호 안은 맞은편 강변 주소다.""" main, _, opposite = addr.partition("(") pick = lambda text: [t for t in re.split(r"\s+", text) if t in DISTRICTS] return pick(main), pick(opposite) def latest_level(obscd): """최근 이틀 시자료 중 마지막 값. 자료가 없으면 (None, None).""" today = datetime.now() rows = get("wl_hrdata", obscd=obscd, startdt=(today - timedelta(days=1)).strftime("%Y%m%d"), enddt=today.strftime("%Y%m%d")).get("list", []) rows = [r for r in rows if r.get("wl") not in (None, "", "-")] if not rows: return None, None last = rows[-1] return float(last["wl"]), last["ymdh"] def main(): stations = [s for s in get("wl_dubwlobs")["list"] if s["obsnm"].startswith("대전시(")] station_rows, region_rows = [], [] for s in stations: info = get("wl_obsinfo", obscd=s["obscd"])["list"][0] level, observed_at = latest_level(s["obscd"]) station_rows.append({ "id": info["wlobscd"], "name": info["obsnm"], "river": info["rivnm"] or "", # 원천 결측은 빈 값으로 둔다 — 추정하지 않는다 "agency": info["mggvcd"], "opened": info["obsopndt"], "basin_area_km2": info["bsnara"], "dist_to_confluence_km": info["rvmjctdis"], "lat": dms_to_decimal(info["lat"]), "lon": dms_to_decimal(info["lon"]), "address": info["addr"].strip(), "latest_level_m": "" if level is None else level, "observed_at": observed_at or "", }) main_side, opposite_side = districts_in(info["addr"]) for d in main_side: region_rows.append({"station_id": info["wlobscd"], "district": d, "side": "본주소"}) for d in opposite_side: region_rows.append({"station_id": info["wlobscd"], "district": d, "side": "맞은편"}) for name, rows in [("stations.csv", station_rows), ("station_regions.csv", region_rows)]: # utf-8-sig: BOM 을 붙여야 엑셀(한국어 Windows)이 한글을 깨뜨리지 않고 연다 with open(DATA / name, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=rows[0].keys()) writer.writeheader() writer.writerows(rows) print(f"{name}: {len(rows)}행") if __name__ == "__main__": main() > python scripts/fetch_wamis.py stations.csv: 19행 station_regions.csv: 26행 5-2. 정리 CSV — 출처 컬럼을 남긴다 하천과 합류 관계는 공식 자료를 보고 직접 옮겨 적었다. 손으로 만든 데이터라 행마다 source 컬럼을 두고, 그래프 관계 속성에도 그대로 넣었다. 나중에 "이 연결은 어디서 온 거지?" 싶을 때 질의로 바로 확인할 수 있다. child,parent,source 갑천,금강,대전시 하천관리사업소 하천현황 유등천,갑천,대전시 하천관리사업소 하천현황 대전천,유등천,대전시 하천관리사업소 하천현황 진잠천,갑천,대전시 3대하천현황 (갑천 지천) 유성천,갑천,대전시 3대하천현황 (갑천 지천) 관평천,갑천,대전시 3대하천현황 (갑천 지천) 매노천,갑천,대전시 3대하천현황 (갑천 지천) 대동천,대전천,위키백과 대동천(대전) 반석천,유성천,위키백과 반석천 rivers.csv 의 등급(국가하천/지방하천)도 출처가 있는 것만 채웠다. 대전시 자료에 국가하천으로 나온 4개와 위키백과에 지방하천으로 나온 2개만 값이 있고, 나머지 4개는 비어 있다. 5-3. Neo4j 실행 — docker-compose.yml # 지식그래프 실습용 Neo4j (로컬 데모 전용 — 비밀번호를 운영 환경에 그대로 쓰지 말 것) services: neo4j: image: neo4j:5-community container_name: kg-neo4j ports: - "7474:7474" # 브라우저 UI http://localhost:7474 - "7687:7687" # Bolt (드라이버·cypher-shell) environment: NEO4J_AUTH: neo4j/kgdemo1234 volumes: - ../data:/import:ro # LOAD CSV 가 읽는 위치 - ../cypher:/cypher:ro # 스크립트 실행용 /import 는 Neo4j가 LOAD CSV ... FROM 'file:///...' 로 읽는 기본 경로라서 CSV 폴더를 여기에 읽기 전용으로 붙였다. 5-4. 스키마 — 유일성 제약조건 // 같은 식별자를 가진 노드가 두 번 만들어지지 않도록 막는다 (ERD의 PK 역할). CREATE CONSTRAINT 행정구역_이름 IF NOT EXISTS FOR (n:행정구역) REQUIRE n.이름 IS UNIQUE; CREATE CONSTRAINT 하천_이름 IF NOT EXISTS FOR (n:하천) REQUIRE n.이름 IS UNIQUE; CREATE CONSTRAINT 관측소_코드 IF NOT EXISTS FOR (n:관측소) REQUIRE n.코드 IS UNIQUE; 제약조건을 걸면 인덱스도 같이 생겨서, 뒤에서 MERGE / MATCH 로 노드를 찾을 때도 빠르다. 5-5. 적재 — LOAD CSV + MERGE MERGE 는 없으면 만들고 있으면 그대로 둔다. 스크립트를 몇 번 다시 돌려도 결과가 같아서 테스트하면서 마음 편하게 재실행할 수 있었다. // --- 노드 --- LOAD CSV WITH HEADERS FROM 'file:///stations.csv' AS row MERGE (s:관측소 {코드: row.id}) SET s.이름 = row.name, s.운영기관 = row.agency, s.관측개시일 = date(row.opened), s.유역면적_km2 = toFloat(row.basin_area_km2), s.합류점거리_km = toFloat(row.dist_to_confluence_km), s.위치 = point({latitude: toFloat(row.lat), longitude: toFloat(row.lon)}), s.주소 = row.address, s.최신수위_m = toFloat(row.latest_level_m), s.관측시각 = row.observed_at; // (행정구역·하천 노드도 같은 방식 — 생략) // --- 관계 --- LOAD CSV WITH HEADERS FROM 'file:///river_tributary.csv' AS row MATCH (child:하천 {이름: row.child}), (parent:하천 {이름: row.parent}) MERGE (child)-[r:지류이다]->(parent) SET r.출처 = row.source; // 하천명이 비어 있는 관측소(원천 결측)는 관계를 만들지 않는다. LOAD CSV WITH HEADERS FROM 'file:///stations.csv' AS row WITH row WHERE row.river IS NOT NULL MATCH (s:관측소 {코드: row.id}), (rv:하천 {이름: row.river}) MERGE (s)-[:관측한다]->(rv); LOAD CSV WITH HEADERS FROM 'file:///station_regions.csv' AS row WITH row WHERE row.side = '본주소' MATCH (s:관측소 {코드: row.station_id}), (r:행정구역 {이름: row.district}) MERGE (s)-[:위치한다]->(r); // 하천이 지나는 구: 관측소 주소(양쪽 강변)에서 이끌어낸다. 관측소가 없는 구간은 반영되지 않는다. LOAD CSV WITH HEADERS FROM 'file:///station_regions.csv' AS row MATCH (:관측소 {코드: row.station_id})-[:관측한다]->(rv:하천), (r:행정구역 {이름: row.district}) MERGE (rv)-[:흐른다]->(r); LOAD CSV 는 빈 칸을 null 로 읽는다. 그래서 WHERE row.river IS NOT NULL 한 줄이면 하천명이 없는 행을 건너뛸 수 있다. 적재가 끝나면 건수부터 맞춰 봤다. MATCH (n) RETURN labels(n)[0] AS 라벨, count(*) AS 개수 ORDER BY 라벨; MATCH ()-[r]->() RETURN type(r) AS 관계, count(*) AS 개수 ORDER BY 관계; 라벨, 개수 관계, 개수 "관측소", 19 "관측한다", 18 ← 관측소 19 − 하천명 결측 1 "하천", 10 "위치한다", 19 "행정구역", 5 "지류이다", 9 "흐른다", 14 관측한다 가 19가 아니라 18인 게 처음엔 이상해 보였는데, 하천명이 비어 있는 관측소 1곳 때문이었다. 이 얘기는 트러블슈팅 D에 정리했다. 5-6. 질의와 결과 Q1. 금강으로 흘러드는 대전 하천 전부 가변 길이 경로 *1.. 을 쓴다. MATCH path = (t:하천)-[:지류이다*1..]->(:하천 {이름: '금강'}) RETURN t.이름 AS 하천, length(path) AS 단계, [n IN nodes(path) | n.이름] AS 경로 ORDER BY 단계, 하천; 하천, 단계, 경로 "갑천", 1, ["갑천", "금강"] "관평천", 2, ["관평천", "갑천", "금강"] "매노천", 2, ["매노천", "갑천", "금강"] "유등천", 2, ["유등천", "갑천", "금강"] "유성천", 2, ["유성천", "갑천", "금강"] "진잠천", 2, ["진잠천", "갑천", "금강"] "대전천", 3, ["대전천", "유등천", "갑천", "금강"] "반석천", 3, ["반석천", "유성천", "갑천", "금강"] "대동천", 4, ["대동천