지난 값이 남지 않는 데이터, 5분마다 직접 쌓기: 서울 실시간 도시데이터 121곳 수집기

서울 실시간 도시데이터를 121곳씩 5분마다 받아 쌓기 시작하자, 한 시간 만에 이상한 점이 네 가지 보였습니다. "실시간" 인구는 받은 시각보다 30분 전 값이었고, 5분마다 기준 시각은 바뀌어도 숫자가 바뀐 경우는 15%뿐이었습니다. 같은 값이 두 번 쌓이기도 했고, 강남역 인구가 예측과 실제 모두 80,000~82,000명인데 예측은 "붐빔", 실제는 "약간 붐빔"으로 라벨이 달랐습니다. 이 글을 따라 하면 지난 값이 남지 않는 데이터를 직접 쌓는 수집기를 만들고, 이런 문제를 쿼리로 먼저 찾아낼 수 있습니다.

이 글은 이런 분께 필요합니다

• 2편에서 지난 파일을 받아 쌓는 백필을 해 본 분
• "실시간 데이터 파이프라인"을 포트폴리오에 넣고 싶은데, 카프카 같은 큰 도구부터 배우기는 부담스러운 분
• API를 한 번 호출해 본 적은 있지만, 같은 호출을 하루 수백 번 안정적으로 돌려 본 적은 없는 분

국민연금 데이터는 한 달에 한 번 바뀌고, 지난 파일이 10년치 남아 있었습니다. 그래서 늦게 시작해도 백필로 따라잡을 수 있었습니다. 서울 실시간 도시데이터는 반대입니다. 서울 주요 121곳의 인구, 혼잡도, 상권 정보를 몇 분 단위로 알려 주지만 지난 값은 API로 다시 받을 수 없습니다. 오늘 오후 3시 강남역 인구를 내일 알고 싶다면, 오늘 오후 3시에 받아 두는 수밖에 없습니다.

Step 1. 인증키 받기, 그리고 sample 키의 함정

서울 열린데이터광장의 오픈 API 주소는 이런 모양입니다.

http://openapi.seoul.go.kr:8088/{인증키}/json/citydata_ppltn/1/5/{장소}

citydata_ppltn은 실시간 인구만 주는 서비스이고, 장소 자리에는 장소 이름(강남역)이나 장소 코드(POI014)를 넣습니다. 인증키 자리에 sample을 넣으면 키 없이도 응답이 옵니다. 문서 예시처럼 따라 해 보기 좋아서 처음엔 이걸로 시작했습니다.

curl "http://openapi.seoul.go.kr:8088/sample/json/citydata_ppltn/1/5/강남역"

{"SeoulRtd.citydata_ppltn":[{"AREA_NM":"광화문·덕수궁","AREA_CD":"POI009", ...

강남역을 요청했는데 광화문·덕수궁이 돌아왔습니다. 홍대 관광특구, POI001로 바꿔 봐도 전부 광화문·덕수궁이었습니다. sample 키는 장소와 상관없이 예시 한 곳만 돌려줍니다. 에러가 나지 않으니, 응답의 AREA_NM을 확인하지 않았다면 121곳 데이터를 모았다고 착각했을 겁니다.

그래서 서울 열린데이터광장에 로그인한 뒤 상단의 "인증키 신청"에서 일반 인증키를 받았습니다. 개인 키로 강남역을 요청하니 그제야 "AREA_NM":"강남역","AREA_CD":"POI014"가 돌아왔습니다. 인증키는 코드에 직접 쓰지 말고 환경변수로 넘깁니다.

한 가지 더 조심할 점이 있습니다. 잘못된 요청을 보내면 서버가 에러 메시지 안에 인증키가 들어간 요청 주소를 그대로 돌려줍니다. 에러 응답을 로그에 남기거나 화면을 캡처해 공유할 때 키가 같이 노출될 수 있습니다. 저도 테스트 중에 한 번 그렇게 됐습니다. 에러 로그는 키를 가린 뒤 저장하세요.

Step 2. 121곳 목록 만들기

121곳은 POI001부터 시작하는 코드로 구분됩니다. 저는 POI001부터 POI140까지 한 번씩 호출해 응답이 오는 코드만 모았습니다.

import json, os, urllib.request, concurrent.futures as cf

KEY = os.environ["SEOUL_API_KEY"]

def probe(i):
    code = f"POI{i:03d}"
    url = f"http://openapi.seoul.go.kr:8088/{KEY}/json/citydata_ppltn/1/5/{code}"
    rows = json.loads(urllib.request.urlopen(url, timeout=30).read()).get("SeoulRtd.citydata_ppltn")
    return (code, rows[0]["AREA_NM"]) if rows else None

with cf.ThreadPoolExecutor(8) as ex:
    places = [p for p in ex.map(probe, range(1, 141)) if p]
json.dump(places, open("places.json", "w", encoding="utf-8"), ensure_ascii=False)
print(len(places), places[-1])

121 ('POI131', '숭례문')

121곳이 나왔고, 코드는 POI131까지 있지만 중간에 빈 번호가 10개(POI022, POI028 등) 있었습니다. 코드가 연속이라고 가정하고 range(1, 122)로 돌렸다면 10곳을 빠뜨렸을 겁니다. 목록은 한 번 만들어 places.json으로 저장해 두고, 가끔 다시 만들어 장소가 추가되거나 빠졌는지 확인합니다.

Step 3. 한 번 실행하면 121곳을 받아 원본 그대로 저장하는 수집기

수집기는 "한 번 실행하면 한 번 수집하고 끝나는" 스크립트로 만듭니다. 반복은 스케줄러에게 맡깁니다. 이렇게 나누면 수집 코드와 반복 주기를 따로 고칠 수 있습니다.

import concurrent.futures as cf, datetime as dt, gzip, json, os, urllib.request

KEY = os.environ["SEOUL_API_KEY"]
PLACES = json.load(open("places.json", encoding="utf-8"))
URL = "http://openapi.seoul.go.kr:8088/{key}/json/citydata_ppltn/1/5/{code}"
KST = dt.timezone(dt.timedelta(hours=9))

def fetch(code):
    try:
        with urllib.request.urlopen(URL.format(key=KEY, code=code), timeout=30) as r:
            return {"code": code, "ok": True, "body": json.loads(r.read())}
    except Exception as e:                      # 실패도 버리지 않고 기록한다
        return {"code": code, "ok": False, "error": type(e).__name__}

now = dt.datetime.now(KST)
with cf.ThreadPoolExecutor(8) as ex:
    results = list(ex.map(fetch, [c for c, _ in PLACES]))
os.makedirs(f"raw/dt={now:%Y-%m-%d}", exist_ok=True)
with gzip.open(f"raw/dt={now:%Y-%m-%d}/{now:%H%M%S}.jsonl.gz", "wt", encoding="utf-8") as f:
    for r in results:
        r["fetched_at"] = now.isoformat(timespec="seconds")
        f.write(json.dumps(r, ensure_ascii=False) + "\n")
print(f"{now:%H:%M:%S} saved {sum(r['ok'] for r in results)}/{len(results)}")

설계에서 신경 쓴 점은 세 가지입니다.

1. 응답을 고치지 않고 통째로 저장합니다. 지금은 인구만 쓰지만, 응답에는 성별·연령대 비율, 상주/비상주 인구 비율, 12시간 예측이 같이 들어 있습니다. 나중에 필요해질 컬럼을 지금 버리면 다시 받을 방법이 없습니다.
2. 실패도 한 줄로 남깁니다. 어느 장소가 언제 실패했는지 알아야, 나중에 비어 있는 칸이 "사람이 없었다"인지 "수집이 안 됐다"인지 구분할 수 있습니다.
3. 실행 한 번에 파일 하나입니다. raw/dt=2026-10-28/160000.jsonl.gz처럼 날짜 폴더와 시각 파일명을 씁니다. 한 번 실행이 중간에 죽어도 다른 시각의 파일에는 영향이 없습니다.

121곳을 8개씩 동시에 요청하면 한 번 수집에 2~4초 정도 걸렸습니다. 파일 하나는 gzip으로 약 15KB이고, 하루 288번 돌면 4MB 남짓입니다.

Step 4. 5분마다 돌리기

스케줄러는 정해진 시간마다 명령을 실행해 주는 프로그램입니다. 맥과 리눅스에는 cron이 기본으로 들어 있어서, 아래 한 줄이면 5분마다 수집기가 돕니다.

*/5 * * * * cd /path/to/city && SEOUL_API_KEY=발급받은키 python3 collect.py >> collect.log 2>&1

저는 오늘 하루만 돌릴 거라 cron 대신 셸 반복문을 썼습니다. 5분 단위 정각(16:00, 16:05, …)에 맞춰 실행되게 하고, 23시 55분에 스스로 멈추게 했습니다.

#!/bin/bash
cd "$(dirname "$0")"
export SEOUL_API_KEY=$(cat .key)
while [ "$(date +%H%M)" -lt 2355 ]; do
  python3 collect.py >> collect.log 2>&1
  sleep $(( 300 - $(date +%s) % 300 ))   # 다음 5분 정각까지 대기
done

nohup caffeinate -i ./loop.sh > /dev/null 2>&1 &

caffeinate -i는 맥이 잠자기에 들어가지 않게 붙잡아 둡니다. 노트북 덮개를 닫거나 잠자기로 들어가면 그 시간 동안 수집기는 멈추고, 그 구간 데이터는 영원히 비게 됩니다. 실제로 하루 이상 돌릴 수집기라면 노트북이 아니라 항상 켜져 있는 서버나 클라우드 스케줄러(GitHub Actions, Airflow 등)로 옮기는 게 맞습니다. 스케줄러를 고르는 기준은 4편 이후 웨어하우스와 함께 다룹니다.

Step 5. 원본을 표로 펴고, 중복부터 지우기

쌓인 JSON 파일을 DuckDB로 한 번에 읽어 표로 펼칩니다.

import duckdb

con = duckdb.connect("city.duckdb")
con.sql("""
CREATE OR REPLACE TABLE stg_population AS
WITH src AS (
  SELECT code, fetched_at::TIMESTAMP AS fetched_at,
         json_extract(body, '$."SeoulRtd.citydata_ppltn"[0]') AS p
  FROM read_json('raw/dt=*/*.jsonl.gz', format = 'newline_delimited',
                 columns = {code: 'VARCHAR', ok: 'BOOLEAN', fetched_at: 'VARCHAR', body: 'JSON'})
  WHERE ok
)
SELECT code, fetched_at,
       p->>'AREA_NM'                                  AS area_name,
       strptime(p->>'PPLTN_TIME', '%Y-%m-%d %H:%M')   AS ppltn_time,
       (p->>'AREA_PPLTN_MIN')::INT                    AS ppltn_min,
       (p->>'AREA_PPLTN_MAX')::INT                    AS ppltn_max,
       p->>'AREA_CONGEST_LVL'                         AS congest_level,
       p->'FCST_PPLTN'                                AS forecast
FROM src
QUALIFY ROW_NUMBER() OVER (PARTITION BY code, ppltn_time ORDER BY fetched_at) = 1
""")

마지막 QUALIFY 줄이 중복 제거입니다. 응답에는 PPLTN_TIME이라는 기준 시각이 들어 있는데, 수집 시각과 기준 시각은 다릅니다. 수집기를 시험하느라 15시 33분과 15시 35분에 연달아 돌렸더니 두 번 모두 기준 시각 15시 05분 데이터가 왔습니다. 같은 장소, 같은 기준 시각의 데이터는 먼저 받은 하나만 남깁니다. 키는 수집 시각이 아니라 **(장소, 기준 시각)**입니다.

print(con.sql("SELECT COUNT(*) AS n, COUNT(DISTINCT code) AS places, MIN(ppltn_time), MAX(ppltn_time) FROM stg_population"))

┌───────┬────────┬─────────────────────┬─────────────────────┐
│   n   │ places │   min(ppltn_time)   │   max(ppltn_time)   │
├───────┼────────┼─────────────────────┼─────────────────────┤
│  1694 │    121 │ 2026-10-28 15:05:00 │ 2026-10-28 16:10:00 │
└───────┴────────┴─────────────────────┴─────────────────────┘

수집 파일 15개, 원본 1,815줄이 중복 제거 후 1,694줄이 됐습니다. 121곳 × 기준 시각 14개입니다. 실패한 요청은 0건이었습니다.

Step 6. 쌓아 보니 보인 것

수집기를 켠 지 약 한 시간, 기준 시각 15:05부터 16:10까지 쌓인 데이터로 확인한 내용입니다. 수집기는 그날 밤까지 계속 돌았습니다.

1) "실시간"은 30분 전입니다

print(con.sql("""
SELECT MIN(date_diff('minute', ppltn_time, fetched_at)) AS min_lag,
       MEDIAN(date_diff('minute', ppltn_time, fetched_at)) AS median_lag,
       MAX(date_diff('minute', ppltn_time, fetched_at)) AS max_lag
FROM stg_population
"""))

┌─────────┬────────────┬─────────┐
│ min_lag │ median_lag │ max_lag │
├─────────┼────────────┼─────────┤
│      28 │       30.0 │      30 │
└─────────┴────────────┴─────────┘

16시 40분에 받은 데이터의 기준 시각은 16시 10분이었습니다. 받은 시각과 기준 시각의 차이는 28~30분으로 일정했습니다. 대시보드에 "지금 강남역 인구"라고 쓰려면 "30분 전 기준"이라고 같이 적어야 합니다. 시간축은 수집 시각이 아니라 기준 시각으로 잡습니다.

2) 5분마다 시각은 바뀌어도 숫자는 15%만 바뀝니다

print(con.sql("""
WITH x AS (
  SELECT *, LAG(ppltn_min) OVER w AS prev_min, LAG(ppltn_max) OVER w AS prev_max
  FROM stg_population WINDOW w AS (PARTITION BY code ORDER BY ppltn_time)
)
SELECT COUNT(*) AS n,
       COUNT(*) FILTER (WHERE ppltn_min <> prev_min OR ppltn_max <> prev_max) AS changed
FROM x WHERE prev_min IS NOT NULL
"""))

┌───────┬─────────┐
│   n   │ changed │
├───────┼─────────┤
│  1573 │     243 │
└───────┴─────────┘

직전 기준 시각과 비교해 인구 값이 바뀐 경우는 1,573번 중 243번(15.4%)이었습니다. 강남역은 15:05부터 16:10까지 14번 내내 80,000~82,000명 그대로였습니다.

이유는 인구가 정확한 숫자가 아니라 범위로 오기 때문입니다. 16시 10분 기준으로 범위 폭이 2,000명인 곳이 62곳, 500명인 곳이 53곳, 100명인 곳이 4곳, 5,000명인 곳이 2곳이었습니다. 인구가 범위 안에서 움직이면 값은 그대로입니다. 그래서 이 데이터로 "5분 사이 몇 % 늘었다"를 계산하면 대부분 0%이다가 가끔 계단처럼 튑니다. 분석할 때는 범위의 가운데 값을 쓰되, 시간 단위로 묶어서 보는 편이 낫습니다.

3) 비어 있는 시간이 있는지 매번 확인합니다

print(con.sql("""
WITH t AS (SELECT DISTINCT ppltn_time FROM stg_population),
g AS (SELECT ppltn_time, LAG(ppltn_time) OVER (ORDER BY ppltn_time) AS prev FROM t)
SELECT COUNT(*) AS times,
       COUNT(*) FILTER (WHERE date_diff('minute', prev, ppltn_time) > 5) AS gaps
FROM g
"""))

┌───────┬───────┐
│ times │ gaps  │
├───────┼───────┤
│    14 │     0 │
└───────┴───────┘

이번에는 공백이 없었습니다. 그래도 이 쿼리는 수집기를 돌리는 동안 매일 확인해야 합니다. 2편의 국민연금 데이터는 빠진 달이 있으면 과거 파일을 다시 받으면 됐지만, 이 데이터는 빠진 5분을 다시 받을 방법이 없습니다. 공백이 생기면 원인(노트북 잠자기, API 장애, 키 만료)을 기록해 두고, 분석할 때 그 구간을 비워 둡니다.

4) 예측과 실제의 인구는 같은데 혼잡도 라벨이 다릅니다

응답의 FCST_PPLTN에는 앞으로 12시간의 시간별 예측이 들어 있습니다. 15:05 기준 응답에 들어 있던 16:00 예측(55분 전 예측)을, 실제 16:00 기준 값과 붙여 봤습니다.

con.sql("""
CREATE OR REPLACE VIEW forecast_vs_actual AS
WITH f AS (
  SELECT s.code, s.area_name, s.ppltn_time AS issued_at,
         strptime(fc->>'FCST_TIME', '%Y-%m-%d %H:%M') AS target_time,
         ((fc->>'FCST_PPLTN_MIN')::INT + (fc->>'FCST_PPLTN_MAX')::INT) / 2.0 AS f_mid,
         fc->>'FCST_CONGEST_LVL' AS f_level
  FROM stg_population s, UNNEST(CAST(s.forecast AS JSON[])) AS t(fc)
)
SELECT f.*, (a.ppltn_min + a.ppltn_max) / 2.0 AS a_mid, a.congest_level AS a_level
FROM f JOIN stg_population a ON a.code = f.code AND a.ppltn_time = f.target_time
""")
print(con.sql("""
SELECT COUNT(*) AS places,
       COUNT(*) FILTER (WHERE f_mid = a_mid) AS same_ppltn,
       COUNT(*) FILTER (WHERE f_mid = a_mid AND f_level = a_level) AS same_ppltn_same_level
FROM forecast_vs_actual
WHERE issued_at = TIMESTAMP '2026-10-28 15:05' AND target_time = TIMESTAMP '2026-10-28 16:00'
"""))

┌────────┬────────────┬───────────────────────┐
│ places │ same_ppltn │ same_ppltn_same_level │
├────────┼────────────┼───────────────────────┤
│    121 │         87 │                    28 │
└────────┴────────────┴───────────────────────┘

121곳 중 87곳은 예측한 인구 범위가 실제와 정확히 같았습니다. 그런데 인구가 같은 87곳 중 혼잡도 라벨까지 같은 곳은 28곳뿐이었고, 52곳은 예측 라벨이 실제보다 한 단계 높았습니다. 강남역은 예측과 실제 모두 80,000~82,000명인데 예측은 "붐빔", 실제는 "약간 붐빔"이었습니다. 여의도, 서울역도 같은 식이었습니다.

예측 라벨과 실제 라벨이 어떤 기준으로 매겨지는지는 응답만으로 알 수 없습니다. 확실한 건 두 라벨을 같은 잣대로 비교하면 안 된다는 점입니다. 예측이 맞았는지는 라벨이 아니라 인구 숫자로 판단합니다. 그리고 이건 55분 전 예측 한 시점의 결과입니다. 수집기를 하루 이상 돌린 뒤에는 예측 시간대(1시간 뒤, 2시간 뒤, …)별로 같은 비교를 해 보세요. 위 forecast_vs_actual 뷰에서 issued_at과 target_time만 바꾸면 됩니다.

막힐 때

증상	원인	해결
어느 장소를 요청해도 광화문·덕수궁만 옴	sample 키는 예시 한 곳만 돌려줌	개인 인증키를 받아 쓴다
에러 메시지에 인증키가 그대로 보임	서버가 요청 주소를 에러에 포함함	로그·캡처를 저장하기 전에 키를 가린다
121곳 중 일부가 빠짐	장소 코드에 빈 번호가 있음	코드를 넉넉히 훑어 실제 응답이 오는 목록을 만든다
같은 데이터가 두 번 쌓임	기준 시각이 바뀌기 전에 다시 수집함	(장소, 기준 시각)으로 중복을 지운다
밤사이 데이터가 비어 있음	노트북이 잠자기에 들어감	caffeinate -i를 쓰거나 항상 켜진 서버로 옮긴다
JSON 경로가 NULL로 나옴	키 이름에 점(.)이 들어 있음	'$."SeoulRtd.citydata_ppltn"[0]'처럼 큰따옴표로 감싼다

재현 자료

수집기 코드와 이 글에서 쓴 수집 결과(15회, 121곳)를 GitHub에 공개해 두었습니다. raw/ 폴더가 들어 있어서 인증키 없이도 flatten.py와 forecast_eval.py로 글의 숫자를 다시 만들 수 있습니다.

다음 편

지금까지 국민연금 데이터는 parquet 파일로, 도시데이터는 DuckDB 파일로 노트북 안에 있습니다. 4편에서는 이 데이터를 어디에 두고 함께 쓸지, 노트북의 DuckDB와 클라우드 웨어하우스(BigQuery 무료 티어)를 비교하며 정합니다.