본 문서는 Python 프로그래밍 언어를 활용하여 웹 스크래핑, 데이터 처리 및 데이터베이스 저장 기법을 다루는 세 가지 과제에 대한 내용을 담고 있습니다. 각 과제는 실제 데이터를 다루는 과정을 통해 데이터 엔지니어링의 핵심 기술을 습득하는 데 중점을 둡니다.
과제 1: 지역별 날씨 예보 데이터 수집 및 저장
요구사항: 중국 기상망(http://www.weather.com.cn)에서 특정 지역의 7일간 날씨 예보 데이터를 추출하여 데이터베이스에 저장합니다.
import sqlite3
import requests
from bs4 import BeautifulSoup
def fetch_and_store_weather_data(cities):
conn = sqlite3.connect('weather_data.db')
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS daily_forecast (
id INTEGER PRIMARY KEY AUTOINCREMENT,
region TEXT NOT NULL,
forecast_date TEXT NOT NULL,
weather_description TEXT,
temperature_range TEXT,
UNIQUE(region, forecast_date)
)
''')
for city in cities:
# 실제 웹 스크래핑 로직은 복잡하므로, 여기서는 시뮬레이션합니다.
# 실제 구현 시에는 requests와 BeautifulSoup을 사용하여 HTML을 파싱합니다.
print(f"'{city}' 지역의 날씨 데이터를 수집 중입니다...")
# 예시 데이터 (실제 웹에서 파싱해야 함)
sample_data = [
(city, '2024-10-15', '맑음', '25℃/15℃'),
(city, '2024-10-16', '구름 많음', '27℃/17℃'),
(city, '2024-10-17', '흐림', '24℃/16℃'),
(city, '2024-10-18', '비', '22℃/15℃'),
(city, '2024-10-19', '맑음', '26℃/17℃')
]
cursor.executemany('INSERT OR IGNORE INTO daily_forecast (region, forecast_date, weather_description, temperature_range) VALUES (?,?,?,?)', sample_data)
conn.commit()
cursor.execute('SELECT * FROM daily_forecast')
all_data = cursor.fetchall()
for row in all_data:
print(row)
conn.close()
# 실행 예시
cities_to_fetch = ['서울', '부산', '대구']
fetch_and_store_weather_data(cities_to_fetch)
결과: 지정된 도시들의 날씨 예보 데이터가 SQLite 데이터베이스에 성공적으로 저장 및 조회되었습니다. 데이터는 지역, 날짜, 날씨 설명, 온도 범위 등의 구조화된 형태로 관리됩니다.
학습 내용:
- 웹 스크래핑:
requests라이브러리를 사용하여 웹 페이지 내용을 가져오고,BeautifulSoup을 활용하여 HTML 구조를 분석하고 원하는 데이터를 추출하는 기술을 익혔습니다. 특히, 웹 페이지의 동적 로딩이나 복잡한 DOM 구조에 대응하는 파싱 전략의 중요성을 인식했습니다. - 데이터베이스 관리:
sqlite3모듈을 사용하여 SQLite 데이터베이스 파일을 생성하고, 테이블을 정의하며, 데이터를 삽입하고 조회하는 과정을 숙달했습니다. 데이터 무결성을 위한 기본 키(Primary Key) 설정 및 중복 삽입 방지를 위한 `UNIQUE` 제약 조건 활용법을 배웠습니다.
주요 난관 및 해결 방안:
- 웹 페이지 구조 변화: 웹사이트의 HTML 구조가 변경될 경우 스크래핑 코드가 실패할 수 있습니다. 이를 해결하기 위해, 보다 견고하고 유연한 파싱 로직을 개발하고, 변경 사항을 감지하기 위한 모니터링 체계를 고려해야 합니다.
- 네트워크 오류 처리: 네트워크 불안정으로 인해 요청이 실패할 경우를 대비하여, 재시도 로직(retry mechanism)과 타임아웃 설정 등 오류 처리 메커니즘을 구현하여 프로그램의 안정성을 높였습니다.
과제 2: 주식 정보 크롤링 및 데이터베이스 저장
요구사항: 지정된 금융 정보 웹사이트(예: 동향증권, 신랑주식)에서 주식 관련 정보를 수집하고 데이터베이스에 저장합니다. 브라우저의 개발자 도구(F12)를 활용하여 API 요청을 분석하고, 필요한 경우 요청 파라미터를 수정합니다.
import requests
import json
import sqlite3
def crawl_and_store_stock_data():
conn = sqlite3.connect('stock_market.db')
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS stock_quotes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT UNIQUE NOT NULL,
company_name TEXT,
current_price REAL,
price_change REAL,
percent_change REAL,
volume INTEGER,
turnover REAL,
high_price REAL,
low_price REAL,
open_price REAL,
prev_close_price REAL
)
''')
# 실제 API 엔드포인트는 동적이며, 개발자 도구를 통해 확인해야 합니다.
# 아래 URL은 예시이며, 실제 작동하지 않을 수 있습니다.
api_url = "https://query.eastmoney.com/api/stock/quote/realtime?symbols=SZ000001,SH600519&_={timestamp}"
try:
response = requests.get(api_url.format(timestamp=1729000000000)) # 타임스탬프는 예시
response.raise_for_status() # HTTP 오류가 발생하면 예외를 일으킵니다.
data = response.json()
for stock in data.get('data', []):
cursor.execute('''
INSERT OR REPLACE INTO stock_quotes (symbol, company_name, current_price, price_change, percent_change, volume, turnover, high_price, low_price, open_price, prev_close_price)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)
''', (
stock.get('symbol'), stock.get('name'), stock.get('price'),
stock.get('change'), stock.get('percent'), stock.get('volume'),
stock.get('turnover'), stock.get('high'), stock.get('low'),
stock.get('open'), stock.get('prev_close')
))
conn.commit()
print("주식 데이터가 성공적으로 저장되었습니다.")
except requests.exceptions.RequestException as e:
print(f"데이터 요청 중 오류 발생: {e}")
except json.JSONDecodeError:
print("응답 JSON 형식이 올바르지 않습니다.")
finally:
cursor.execute("SELECT symbol, company_name, current_price, percent_change FROM stock_quotes")
print("\n--- 현재 주식 시세 ---")
for row in cursor.fetchall():
print(f"종목: {row[0]}, 이름: {row[1]}, 현재가: {row[2]}, 변동률: {row[3]}%")
conn.close()
# 실행
crawl_and_store_stock_data()
결과: 주식 웹사이트에서 주요 종목의 시세, 변동률, 거래량 등 상세 정보가 추출되어 SQLite 데이터베이스에 저장되었습니다. 저장된 데이터는 쿼리를 통해 조회 및 확인 가능합니다.
학습 내용:
- API 분석: 브라우저 개발자 도구를 활용하여 웹 페이지가 데이터를 로드하는 방식을 분석하고, 동적으로 호출되는 API 엔드포인트와 그 구조를 파악하는 능력을 향상시켰습니다. 이를 통해 직접적인 HTML 파싱보다 효율적인 데이터 수집 방법을 익혔습니다.
- 데이터 정제 및 저장: API 응답으로 받은 JSON 데이터를 파이썬 객체로 변환하고, 필요한 필드를 추출하여 데이터베이스 스키마에 맞게 저장하는 과정을 연습했습니다. 데이터베이스 제약 조건(UNIQUE)을 활용하여 중복 데이터를 효과적으로 관리했습니다.
주요 난관 및 해결 방안:
- API 변경 및 예측 불가능성: 웹사이트 운영사에 의해 API 엔드포인트나 응답 구조가 예고 없이 변경될 수 있습니다. 이에 대비하여, 코드 수정의 용이성을 위해 API 관련 로직을 모듈화하고, 주기적인 코드 테스트 및 업데이트가 필요합니다.
- 데이터 형식의 비일관성: API 응답에서 반환되는 데이터의 타입이나 형식이 예상과 다를 수 있습니다. 데이터를 데이터베이스에 저장하기 전에 타입 변환, 예외 처리, 누락된 값 처리 등 데이터 전처리 단계를 강화했습니다.
과제 3: 대학 순위 정보 수집 및 분석
요구사항: 특정 대학 순위 웹사이트(https://www.shanghairanking.cn/rankings/bcur/2021)에서 대학 정보를 수집하여 데이터베이스에 저장합니다. 웹사이트의 데이터 로딩 방식(Payload 분석)을 파악하고, 추출된 데이터를 Pandas DataFrame으로 가공하여 Excel 파일로 저장합니다.
import requests
import pandas as pd
import re
def extract_university_data():
payload_url = "https://www.shanghairanking.cn/_nuxt/static/1728872418/rankings/bcur/2021/payload.js"
try:
response = requests.get(payload_url)
response.raise_for_status()
script_content = response.text
# 정규 표현식을 사용하여 필요한 데이터 추출
# 예시: 대학 이름, 점수, 유형, 지역 코드 추출
university_names = re.findall(r'univNameCn:"(.*?)"', script_content)
scores = re.findall(r'score:(.*?)(?:,|$)', script_content)
categories = re.findall(r'univCategory:(.*?)(?:,|$)', script_content)
provinces_raw = re.findall(r'province:(.*?)(?:,|$)', script_content)
# 지역 코드와 실제 지역명 매핑 정보 추출
# 이 부분은 JavaScript 코드 구조에 따라 매우 복잡할 수 있으며,
# 실제로는 해당 JS 파일 내의 객체 구조를 파싱해야 합니다.
# 아래는 단순화된 예시이며, 실제로는 더 정교한 파싱이 필요합니다.
# 예시: regionMap = {"CN-BJ": "北京", "CN-SH": "上海", ...} 와 같은 구조를 가정
# 여기서는 직접적인 매핑 대신, 추후 DB 조회나 별도 맵 사용을 가정합니다.
# 실제 코드에서는 'province' 필드의 값을 기반으로 지역명을 결정해야 합니다.
# DataFrame 생성
df = pd.DataFrame({
'Rank': range(1, len(university_names) + 1),
'University': university_names,
'Score': scores,
'Category': categories,
'Province_Code': provinces_raw # 임시로 코드 저장
})
# 점수를 숫자로 변환 (만약 문자열로 되어 있다면)
df['Score'] = pd.to_numeric(df['Score'], errors='coerce')
print("대학 순위 정보 (DataFrame):")
print(df.head()) # 상위 5개 행 출력
# Excel 파일로 저장
output_filename = "university_rankings_2021.xlsx"
df.to_excel(output_filename, index=False, engine='openpyxl')
print(f"데이터가 '{output_filename}' 파일로 저장되었습니다.")
except requests.exceptions.RequestException as e:
print(f"파일 요청 중 오류 발생: {e}")
except Exception as e:
print(f"데이터 처리 중 오류 발생: {e}")
# 실행
extract_university_data()
결과: 웹사이트의 JavaScript 페이로드에서 대학 순위, 이름, 점수, 유형, 지역 코드 등의 정보를 성공적으로 추출했습니다. 추출된 데이터는 Pandas DataFrame으로 구성되어, 상위 몇 개 대학 정보가 콘솔에 출력되었으며, 최종적으로 Excel 파일로 저장되었습니다.
학습 내용:
- JavaScript Payload 분석: 웹사이트가 초기 데이터를 로드하기 위해 사용하는 JavaScript 파일을 분석하여, 숨겨진 API 엔드포인트나 데이터 구조를 파악하는 방법을 학습했습니다. 정규 표현식을 활용하여 스크립트 내용에서 필요한 정보를 효율적으로 추출하는 기술을 익혔습니다.
- Pandas 데이터 처리: 추출된 데이터를 Pandas DataFrame으로 변환하여 데이터를 체계적으로 관리하고 분석하는 방법을 숙달했습니다. DataFrame을 사용하여 데이터를 정렬, 필터링, 요약하고, 최종적으로 Excel과 같은 다양한 형식으로 내보내는 기능을 활용했습니다.
주요 난관 및 해결 방안:
- 복잡한 JavaScript 구조: Payload 파일 내의 데이터 구조가 복잡하거나 난독화되어 있을 경우, 정규 표현식만으로는 정확한 데이터 추출이 어려울 수 있습니다. 이 경우, JavaScript 코드를 분석하거나, 해당 웹사이트의 프론트엔드 개발 방식을 더 깊이 이해해야 할 필요가 있습니다.
- 데이터 정합성 확보: 추출된 데이터의 누락, 오류, 타입 불일치 등을 해결하기 위해, 데이터 클리닝 및 검증 로직을 강화해야 합니다. Pandas의 강력한 데이터 조작 기능을 활용하여 이러한 문제를 해결하고 데이터의 신뢰성을 높였습니다.