파이썬 API 데이터 수집 완벽 가이드 | 공공데이터부터 소셜미디어까지 [시리즈 2탄]
![파이썬 API 데이터 수집 완벽 가이드 | 공공데이터부터 소셜미디어까지 [시리즈 2탄]](https://images.unsplash.com/photo-1551288049-bebda4e38f71?q=80&auto=format&fit=crop&ixlib=rb-4.0.3&w=1200)
안녕하세요! 데이터 수집 시리즈 1탄에서 웹 스크래핑의 기초를 함께 배워보셨는데요, 많은 분들이 "API는 어떻게 사용하나요?"라는 질문을 주셨습니다. 오늘은 그 답을 드리는 시간입니다!
1탄에서 BeautifulSoup으로 뉴스 데이터를 수집해보셨다면, 2탄에서는 한 단계 더 나아가 REST API를 활용한 전문적인 데이터 수집 방법을 배워보겠습니다.
🔍 웹 스크래핑 vs API, 어떤 차이가 있을까요?
1탄에서 배운 웹 스크래핑과 API 활용은 어떤 차이가 있을까요? 실제 업무에서 겪어보신 분들은 아시겠지만, 각각 장단점이 명확합니다.
웹 스크래핑의 특징
# 1탄에서 사용한 방식
response = requests.get(url)
soup = BeautifulSoup(response.content, 'html.parser')
titles = soup.select('.news_title') # HTML 구조 의존적
장점: 모든 웹사이트에 적용 가능, 시각적으로 보이는 데이터 수집
단점: HTML 구조 변경 시 코드 수정 필요, 속도 제한, 법적 이슈 가능성
API 활용의 특징
# 2탄에서 배울 방식
response = requests.get(api_url, params=params, headers=headers)
data = response.json() # 구조화된 JSON 데이터
장점: 안정적인 데이터 구조, 빠른 속도, 공식 지원
단점: API 제공 사이트만 가능, 이용 제한 존재
제 경험으로는 공식 API가 있다면 무조건 API를 먼저 고려하시는 것을 추천드립니다. 안정성과 속도 면에서 확실한 차이를 느끼실 수 있거든요.
📚 REST API 기초, 이것만 알면 됩니다
REST API를 처음 접하시는 분들을 위해 핵심 개념부터 차근차근 설명드리겠습니다.
REST API란?
REST(Representational State Transfer) API는 웹 서비스 간의 통신 방식 중 하나입니다. 마치 레스토랑의 웨이터같은 역할을 한다고 생각하시면 됩니다.
- 고객(클라이언트): 데이터가 필요한 우리
- 웨이터(API): 요청을 받아 주방에 전달하고 결과를 가져다주는 역할
- 주방(서버): 실제 데이터를 처리하고 관리하는 곳
HTTP 메서드 이해하기
# GET: 데이터 조회 (가장 많이 사용)
response = requests.get('https://api.example.com/users')
# POST: 새 데이터 생성
response = requests.post('https://api.example.com/users', json=new_user_data)
# PUT: 기존 데이터 전체 수정
response = requests.put('https://api.example.com/users/123', json=updated_data)
# DELETE: 데이터 삭제
response = requests.delete('https://api.example.com/users/123')
데이터 수집에서는 주로 GET 메서드를 사용하게 됩니다. 나머지는 데이터를 조작할 때 필요한 메서드들이에요.
JSON 데이터 구조 이해하기
API는 대부분 JSON(JavaScript Object Notation) 형태로 데이터를 제공합니다. Python의 딕셔너리와 매우 유사한 구조입니다.
# 기본적인 JSON 구조
weather_data = {
"location": "서울",
"temperature": 23.5,
"humidity": 65,
"forecasts": [
{"date": "2025-01-20", "temp": 25.0},
{"date": "2025-01-21", "temp": 22.3}
]
}
# Python에서 JSON 데이터 접근
location = weather_data["location"] # "서울"
first_forecast = weather_data["forecasts"][0] # 첫 번째 예보 데이터
🌤️ 실습: 기상청 공공데이터 API로 날씨 수집하기
이제 실제로 공공데이터포털의 기상청 API를 활용해서 전국 날씨 데이터를 수집해보겠습니다. 1탄의 NewsCollector와 유사한 구조로 WeatherCollector 클래스를 만들어보세요.
1단계: 공공데이터포털 API 키 발급받기
먼저 공공데이터포털에서 회원가입 후 "기상청_단기예보 조회서비스"를 신청하세요.
import requests
import pandas as pd
import os
from datetime import datetime
class WeatherCollector:
"""기상청 공공데이터 API를 활용한 날씨 데이터 수집기"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0"
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
# 주요 도시의 격자 좌표 (기상청 API는 격자 좌표 시스템 사용)
self.regions = {
"서울": {"nx": 60, "ny": 127},
"부산": {"nx": 98, "ny": 76},
"대구": {"nx": 89, "ny": 90},
"인천": {"nx": 55, "ny": 124},
"광주": {"nx": 58, "ny": 74},
"대전": {"nx": 67, "ny": 100}
}
2단계: API 요청 함수 구현
def get_weather_forecast(self, region: str) -> dict:
"""특정 지역의 날씨 예보 데이터 수집"""
try:
if region not in self.regions:
raise ValueError(f"지원하지 않는 지역: {region}")
# 현재 시간 기준으로 예보 시간 계산
now = datetime.now()
base_date = now.strftime("%Y%m%d")
base_time = self._get_forecast_time(now)
params = {
'serviceKey': self.api_key,
'pageNo': '1',
'numOfRows': '1000',
'dataType': 'JSON',
'base_date': base_date,
'base_time': base_time,
'nx': self.regions[region]['nx'],
'ny': self.regions[region]['ny']
}
url = f"{self.base_url}/getVilageFcst"
response = requests.get(url, params=params, headers=self.headers)
response.raise_for_status() # HTTP 에러 시 예외 발생
data = response.json()
return self._parse_weather_data(data, region)
except requests.exceptions.RequestException as e:
print(f"API 요청 실패 ({region}): {e}")
return {}
except Exception as e:
print(f"데이터 처리 실패 ({region}): {e}")
return {}
3단계: JSON 데이터 파싱 고급 기법
기상청 API는 복잡한 구조의 JSON을 제공합니다. 이를 효과적으로 처리하는 방법을 알아보세요.
def _parse_weather_data(self, data: dict, region: str) -> dict:
"""복잡한 기상청 JSON 데이터를 구조화된 형태로 변환"""
try:
items = data['response']['body']['items']['item']
weather_info = {
'region': region,
'collected_at': datetime.now().isoformat(),
'forecasts': []
}
# 날짜/시간별로 데이터 그룹화
forecast_dict = {}
for item in items:
date_time = f"{item['fcstDate']}_{item['fcstTime']}"
if date_time not in forecast_dict:
forecast_dict[date_time] = {
'date': item['fcstDate'],
'time': item['fcstTime'],
'data': {}
}
# 기상 요소별 데이터 저장
category = item['category']
value = item['fcstValue']
forecast_dict[date_time]['data'][category] = value
# 사용자 친화적 형태로 변환
for key, forecast in forecast_dict.items():
data_dict = forecast['data']
forecast_info = {
'date': forecast['date'],
'time': forecast['time'],
'temperature': data_dict.get('TMP', 'N/A'), # 기온
'humidity': data_dict.get('REH', 'N/A'), # 습도
'precipitation': data_dict.get('PCP', 'N/A'), # 강수량
'wind_speed': data_dict.get('WSD', 'N/A'), # 풍속
'sky_condition': self._get_sky_condition(data_dict.get('SKY', '1'))
}
weather_info['forecasts'].append(forecast_info)
return weather_info
except KeyError as e:
print(f"데이터 파싱 오류 ({region}): 필요한 키가 없습니다 - {e}")
return {}
🔐 API 키 관리와 보안, 이렇게 하세요
API 키는 여러분의 계정과 직결된 중요한 정보입니다. 절대 코드에 직접 입력하지 마세요!
환경변수 활용법
import os
from dotenv import load_dotenv
# .env 파일 로드
load_dotenv()
# 환경변수에서 API 키 가져오기
API_KEY = os.getenv('WEATHER_API_KEY')
if not API_KEY:
raise ValueError("WEATHER_API_KEY 환경변수를 설정해주세요.")
collector = WeatherCollector(API_KEY)
.env 파일 생성 (프로젝트 루트 폴더에)
WEATHER_API_KEY=your_actual_api_key_here
요청 제한 처리하기
대부분의 API는 시간당 요청 횟수를 제한합니다. 이를 적절히 처리해야 안정적인 데이터 수집이 가능합니다.
import time
from functools import wraps
def rate_limit(calls_per_second=1):
"""API 요청 속도 제한 데코레이터"""
min_interval = 1.0 / calls_per_second
last_called = [0.0]
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
left_to_wait = min_interval - elapsed
if left_to_wait > 0:
time.sleep(left_to_wait)
ret = func(*args, **kwargs)
last_called[0] = time.time()
return ret
return wrapper
return decorator
# 사용 예시
@rate_limit(calls_per_second=0.5) # 2초에 1번 요청
def safe_api_request(self, region):
return self.get_weather_forecast(region)
⚡ 비동기 처리로 수집 속도 10배 향상시키기
여러 지역의 데이터를 동시에 수집할 때는 비동기 처리가 필수입니다. 순차적으로 8개 지역을 요청하면 16초가 걸리지만, 비동기로 처리하면 2초 만에 완료됩니다!
asyncio와 aiohttp 활용
import asyncio
import aiohttp
async def get_weather_async(self, session: aiohttp.ClientSession, region: str) -> dict:
"""비동기 날씨 데이터 수집"""
try:
now = datetime.now()
base_date = now.strftime("%Y%m%d")
base_time = self._get_forecast_time(now)
params = {
'serviceKey': self.api_key,
'pageNo': '1',
'numOfRows': '1000',
'dataType': 'JSON',
'base_date': base_date,
'base_time': base_time,
'nx': self.regions[region]['nx'],
'ny': self.regions[region]['ny']
}
url = f"{self.base_url}/getVilageFcst"
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
return self._parse_weather_data(data, region)
else:
print(f"API 응답 오류 ({region}): {response.status}")
return {}
except Exception as e:
print(f"비동기 요청 실패 ({region}): {e}")
return {}
async def collect_all_regions_async(self) -> list:
"""모든 지역 날씨 데이터 비동기 수집"""
async with aiohttp.ClientSession() as session:
tasks = []
for region in self.regions.keys():
task = self.get_weather_async(session, region)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
# 성공한 결과만 필터링
valid_results = []
for result in results:
if isinstance(result, dict) and result:
valid_results.append(result)
return valid_results
실행 및 성능 비교
import time
# 사용 예시
if __name__ == "__main__":
API_KEY = os.getenv('WEATHER_API_KEY')
collector = WeatherCollector(API_KEY)
print("=== 비동기 수집 시작 ===")
start_time = time.time()
# 비동기 실행
all_weather_data = asyncio.run(collector.collect_all_regions_async())
end_time = time.time()
print(f"수집 완료! 소요시간: {end_time - start_time:.2f}초")
print(f"수집된 지역: {len(all_weather_data)}개")
💾 데이터 저장 및 활용 방법
수집한 데이터는 다양한 형태로 저장할 수 있습니다. 용도에 따라 적절한 형식을 선택하세요.
CSV 형태로 저장 (분석용)
def save_to_csv(self, weather_data: list, filename: str = None) -> str:
"""수집된 데이터를 CSV 파일로 저장"""
if not weather_data:
print("저장할 데이터가 없습니다.")
return ""
# 데이터를 평면화하여 DataFrame 생성
flattened_data = []
for region_data in weather_data:
region = region_data.get('region', 'Unknown')
collected_at = region_data.get('collected_at', '')
for forecast in region_data.get('forecasts', []):
row = {
'region': region,
'collected_at': collected_at,
**forecast # 예보 데이터 모든 필드 포함
}
flattened_data.append(row)
df = pd.DataFrame(flattened_data)
if filename is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"weather_data_{timestamp}.csv"
df.to_csv(filename, index=False, encoding='utf-8-sig')
print(f"데이터가 {filename}에 저장되었습니다. (총 {len(df)}개 레코드)")
return filename
JSON 형태로 저장 (구조 보존용)
import json
def save_to_json(self, weather_data: list, filename: str = None) -> str:
"""수집된 데이터를 JSON 파일로 저장"""
if filename is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"weather_data_{timestamp}.json"
with open(filename, 'w', encoding='utf-8') as f:
json.dump(weather_data, f, ensure_ascii=False, indent=2)
print(f"데이터가 {filename}에 저장되었습니다.")
return filename
📊 수집 데이터 통계 및 검증
데이터 수집 후에는 반드시 품질을 확인해야 합니다.
def get_summary_stats(self, weather_data: list) -> dict:
"""수집된 데이터의 요약 통계"""
if not weather_data:
return {}
total_forecasts = sum(len(region['forecasts']) for region in weather_data)
regions = [region['region'] for region in weather_data]
# 온도 통계 (숫자 값만 처리)
temperatures = []
for region_data in weather_data:
for forecast in region_data['forecasts']:
try:
temp = float(forecast['temperature'])
temperatures.append(temp)
except (ValueError, TypeError):
continue
stats = {
'total_regions': len(regions),
'regions': regions,
'total_forecasts': total_forecasts,
'data_collection_time': datetime.now().isoformat()
}
if temperatures:
stats.update({
'temperature_stats': {
'min': min(temperatures),
'max': max(temperatures),
'avg': round(sum(temperatures) / len(temperatures), 1)
}
})
return stats
🚀 실전 활용 예제
이제 완성된 WeatherCollector를 실제로 사용해보겠습니다.
# 실행 예제
def main():
# 환경변수에서 API 키 가져오기
API_KEY = os.getenv('WEATHER_API_KEY')
if not API_KEY:
print("WEATHER_API_KEY 환경변수를 설정해주세요.")
return
# WeatherCollector 인스턴스 생성
collector = WeatherCollector(API_KEY)
print("=== 날씨 데이터 수집 시작 ===")
# 전국 날씨 데이터 비동기 수집
start_time = time.time()
all_weather_data = asyncio.run(collector.collect_all_regions_async())
end_time = time.time()
print(f"수집 완료! 소요시간: {end_time - start_time:.2f}초")
if all_weather_data:
# 데이터 저장
csv_file = collector.save_to_csv(all_weather_data)
json_file = collector.save_to_json(all_weather_data)
# 통계 정보 출력
stats = collector.get_summary_stats(all_weather_data)
print(f"\n=== 수집 통계 ===")
print(f"총 지역: {stats.get('total_regions', 0)}")
print(f"총 예보 데이터: {stats.get('total_forecasts', 0)}")
temp_stats = stats.get('temperature_stats')
if temp_stats:
print(f"온도 범위: {temp_stats['min']}°C ~ {temp_stats['max']}°C")
print(f"평균 온도: {temp_stats['avg']}°C")
else:
print("수집된 데이터가 없습니다. API 키와 네트워크 연결을 확인해주세요.")
if __name__ == "__main__":
main()
🎯 다음 단계와 주의사항
꼭 지켜야 할 API 사용 원칙
- 이용약관 준수: 각 API의 이용약관을 반드시 확인하세요
- 요청 제한 지키기: 무분별한 요청은 API 키 차단으로 이어집니다
- 에러 처리: 네트워크 오류, API 오류 등을 적절히 처리하세요
- 데이터 검증: 수집된 데이터가 예상한 형태인지 항상 확인하세요
API 선택 시 고려사항
- 무료 vs 유료: 데이터 양과 빈도에 따른 비용 고려
- 응답 속도: 실시간성이 중요한 경우 응답 시간 확인
- 데이터 품질: 정확성과 최신성 검증
- 지원 범위: 필요한 지역이나 기간의 데이터 제공 여부
🔮 시리즈 3탄 예고
2탄에서 API 활용 데이터 수집의 기초를 마스터하셨다면, 3탄에서는 더욱 고급 기능들을 다뤄보겠습니다:
📝 시리즈 3탄 주요 내용
- 데이터베이스 연동: 수집한 데이터를 SQLite, MongoDB에 체계적으로 저장
- 스케줄링 자동화: cron, APScheduler를 활용한 정기적 데이터 수집
- 실시간 모니터링: 수집 상태 알림, 로그 관리 시스템
- 클라우드 배포: AWS, GCP에서 안정적인 데이터 수집 서비스 구축
- 대용량 처리: 멀티프로세싱, 분산 처리로 대규모 데이터 처리
실습 프로젝트: 금융 데이터 + 날씨 데이터를 결합한 투자 인사이트 시스템 구축
💬 마무리
오늘 2탄에서는 웹 스크래핑에서 한 단계 발전한 API 활용 데이터 수집을 배워보셨습니다. 특히 공공데이터포털의 기상청 API를 통해 실제 프로젝트에서 사용할 수 있는 수준의 코드를 작성해보셨어요.
API 활용의 핵심은 안정성과 효율성입니다. 오늘 배운 비동기 처리, 에러 핸들링, 데이터 검증 기법들을 실제 프로젝트에 적용해보시길 바랍니다.
궁금한 점이나 어려운 부분이 있으시면 댓글로 공유해주세요. 여러분의 피드백이 더 나은 시리즈를 만들어갑니다!
다음 3탄에서는 수집한 데이터를 어떻게 체계적으로 관리하고 자동화할 수 있는지 알아보겠습니다. 기대해주세요! 🚀
📌 이 글이 도움이 되셨다면
- 👍 좋아요로 응원해주세요
- 💬 댓글로 의견을 나눠주세요
- 📤 동료들과 공유해주세요
🔗 관련 링크
Featured image: 데이터 수집과 API 연동을 상징하는 이미지