토양환경 공공데이터 활용 가이드 및 파이썬 REST API 연동 실무

부지 개발이나 환경 실사(Due Diligence)를 수행하는 실무자가 가장 먼저 직면하는 과제는 대상 부지 주변의 토양·지하수 환경 기초 데이터를 어디서, 어떤 형식으로 수집하고 검증할 것인가입니다. 기후에너지환경부와 한국환경공단이 공공데이터포털을 통해 개방하는 국가 토양측정망과 지하수 관측망 데이터는 부지의 배경 오염도를 파악하는 1차 스크리닝 도구로 유용하지만, 고정 지점 데이터라는 태생적 한계가 존재합니다. 공공데이터 API의 데이터셋 구조, 결측치·불검출(검출한계 미만) 처리 기준, 좌표계 변환 및 파이썬 연동 파이프라인의 실무 구축 방안을 객관적으로 해설합니다.

환경 공공데이터 목적별 추천 데이터셋 및 분석 한계 비교

공공 환경데이터는 수집 기관과 목적에 따라 데이터의 성격이 상이하므로, 실무 분석 목적에 맞추어 적절한 데이터셋을 조합해야 합니다.

파이썬 기반 공공데이터 수집 및 전처리 파이프라인

공공데이터포털 REST API를 연동하여 환경 분석 파이프라인을 구축할 때 마주치는 3대 기술 과제는 다음과 같습니다.

• 1. 결측치 및 정량한계 미만(검출한계 미만) 처리: 환경 분석에서 "불검출"은 0이 아닌 검출한계(Method Detection Limit, MDL) 미만을 의미합니다. 통계 분석 시에는 단순 0으로 치환하기보다 환경 통계 기준에 따라 검출한계의 1/2 또는 1/√2를 대입하는 대체법을 고려하거나, 목적에 따라 결측치로 분리 처리합니다.
• 2. 좌표계 변환 및 공간 조인(Spatial Join): 공공데이터포털의 위치 정보는 WGS84 위경도(EPSG:4326)로 제공되는 반면, 국내 토지정보시스템 및 수치지도 레이어는 한국표준 직각좌표계(GRS80 / EPSG:5179)를 주로 사용합니다. GeoPandas의 to_crs(epsg=5179)를 사용하여 좌표계를 통일한 뒤 부지 경계 버퍼와의 공간 중첩을 분석합니다.
• 3. API 트래픽 제어 및 로컬 캐싱: 호출 한도·페이지 크기·재시도 간격은 데이터셋별 서비스 명세와 승인 트래픽에 따릅니다. 아래 예제의 0.2초 대기는 설명용 가정이며 허용 호출률이나 응답시간 보장이 아닙니다. 대규모 배치 수집 시에는 승인 조건에 맞는 대기·재시도 정책을 두고 로컬 데이터베이스(SQLite / PostgreSQL)에 원본 응답을 캐싱하여 중복 호출을 방지합니다.

파이썬 API 호출 표준 연동 구조 (의사코드 예제)

다음 코드는 공공데이터포털의 표준 REST API 연동 및 페이징 수집 아키텍처를 설명하기 위한 파이썬 의사 예제(Pseudocode)입니다. 실제 호출 시에는 공공데이터포털에서 발급받은 일반 인증키(Encoding/Decoding)와 활성화된 상세 엔드포인트 URL을 확인하여 적용해야 합니다.

# [표준 연동 구조 의사코드 (Pseudocode Architecture)]
# ※ 본 예제는 공공데이터포털 환경 API의 파이썬 연동 표준 절차(인증, 페이징, 예외 처리)를 안내합니다.
# 실제 운영 시에는 공공데이터포털(data.go.kr)의 최신 서비스 엔드포인트 및 승인된 서비스키를 대입하십시오.

import requests
import pandas as pd
import time

def fetch_soil_quality_data(service_key: str, base_year: str, max_pages: int = 5) -> pd.DataFrame:
    """
    공공데이터포털 환경측정망 API 표준 페이징 수집 함수 (의사코드 예시)
    """
    endpoint = "https://apis.data.go.kr/1482249/SoilQualityService/getSoilQualityList"
    all_records = []
    
    headers = {"Accept": "application/json"}
    
    for page in range(1, max_pages + 1):
        params = {
            "serviceKey": service_key,
            "pageNo": str(page),
            "numOfRows": "100",
            "resultType": "json",
            "year": base_year
        }
        
        try:
            response = requests.get(endpoint, params=params, headers=headers, timeout=10)
            if response.status_code != 200:
                print(f"[경고] HTTP 오류 코드 수신: {response.status_code}")
                break
                
            data = response.json()
            body = data.get("response", {}).get("body", {})
            items = body.get("items", [])
            
            if not items:
                print(f"[{page}페이지] 추가 데이터 없음. 수집 종료.")
                break
                
            all_records.extend(items)
            time.sleep(0.2)  # 설명용 가정값: 실제 간격은 API 승인 조건·재시도 정책에 맞춰 설정
            
        except requests.exceptions.RequestException as e:
            print(f"[네트워크 예외 발생]: {e}")
            break
            
    df = pd.DataFrame(all_records)
    print(f"총 {len(df)}건의 측정망 레코드가 데이터프레임으로 수집되었습니다.")
    return df

# [실행 예시 가이드]
# if __name__ == "__main__":
#     MY_KEY = "발급받은_디코딩_서비스키"
#     df_soil = fetch_soil_quality_data(service_key=MY_KEY, base_year="2025")

공공데이터 이용 시 라이선스 및 법적 유의사항

자주 묻는 질문