데이터 수집

데이터 수집

1) 학습개요 요약

• 데이터 분석의 출발점은 목적 적합성을 갖춘 데이터 확보입니다.
• 단순 수집을 넘어, 정확성·완전성·일관성 등 품질 요건을 만족하도록 구조와 절차를 설계해야 합니다.
• 정형/반정형/비정형 유형을 구분하고, 파일·API·웹 스크래핑 등 상황에 맞는 수집 전략을 선택합니다.

2) 학습목표 매핑

• 목표 1: 왜 수집이 필요한가 → 가치·의사결정·경쟁력, 수집 정의 및 품질의 중요성.
• 목표 2: 데이터 유형 구분 → 정형/반정형/비정형의 구조·특성·한계.
• 목표 3: 다양한 소스 수집 코드 → 파일(pandas), API(requests/GraphQL/WebSocket), 스크래핑(requests+BeautifulSoup/Selenium) 예시.

---

3) 데이터 수집의 이해

3.1 왜 필요한가

• 의사결정 품질은 데이터 품질과 직결됩니다(개인·기업·국가 경쟁력에 영향).
• “수집”은 모으기가 아니라 분석 목적에 맞게 준비(형식·주기·정확성 보장)하는 과정입니다.

3.2 데이터 품질 핵심(ISO 8000 관점)

• 정확성: 사실과 일치
• 완전성: 결측(누락) 없음
• 일관성: 포맷/규칙의 균일성
• 유효성: 정의된 규칙/범위에 부합
• 적시성: 최신성 및 갱신 주기 충족
• 상호운용성: 시스템 간 호환

3.3 수집 시 흔한 어려움

• 양: 충분한 표본/커버리지 확보
• 다양성: 출처·스키마·포맷의 이질성
• 정확성/일관성: 중복, 포맷 불일치, 인코딩/로캘 문제

---

4) 데이터의 유형

4.1 정형(Structured)

• 특징: 고정 스키마, 행·열 테이블. RDBMS에 적합, 질의·집계가 용이.
• 한계: 예측하지 못한 구조/속성의 유연성 낮음.

4.2 비정형(Unstructured)

• 특징: 사전 스키마 없음(문서, 이미지, 오디오/비디오, 자유 텍스트 등).
• 의의: 생산되는 데이터의 80%가 비정형 데이터.
• 맥락 정보가 풍부하나, 저장 및 관리가 어렵고, 처리 난이도 높음(NLP, CV 등 필요).

4.3 반정형(Semi-structured)

• 특징: 키-값/태그/계층 구조(JSON, XML, 로그, 이메일 헤더 등).
• 장점: 유연한 스키마 진화, NoSQL/데이터 레이크에 적합.
• 주의: 분석 전 스키마 추론·정제 필요.

---

5) 데이터 수집 방법

5.1 파일 기반 수집(일회성/배치에 적합)

• 대표 포맷: CSV, Excel, JSON, XML, HTML, TXT, LOG.
• 권장 절차: 인코딩 확인 → 스키마 정의/캐스팅 → 결측/이상치 처리 → 검증/샘플링.

# CSV/Excel/JSON 빠른 로드 예시 (Python + pandas)
import pandas as pd

df_csv = pd.read_csv("data.csv")                    # 기본: 콤마 구분, UTF-8 가정
df_xls = pd.read_excel("report.xlsx", sheet_name=0) # 시트 지정
df_json = pd.read_json("data.json", lines=False)    # JSON Lines면 lines=True

# 타입 캐스팅 & 기본 정제
df = df_csv.rename(columns=lambda c: c.strip())
df["date"] = pd.to_datetime(df["date"], errors="coerce")
df = df.dropna(subset=["id"])                       # 필수키 결측 제거

5.2 API 기반 수집(자동화/최신성/접근 제어)

• REST API(HTTP 요청/응답, JSON/XML), GraphQL(필요한 필드만 질의), WebSocket(양방향 실시간).
• 핵심 체크: 인증(키/토큰), Rate limit, 페이징(cursor/offset), 재시도(백오프), 스키마 버전.

# REST API 예시 (requests)
import requests

BASE = "https://api.example.com/v1/items"
params = {"limit": 100, "page": 1}
headers = {"Authorization": "Bearer "}

items = []
while True:
    r = requests.get(BASE, params=params, headers=headers, timeout=10)
    r.raise_for_status()
    data = r.json()
    items.extend(data["results"])
    if not data.get("next"): break
    params["page"] += 1

# GraphQL 예시: 필요한 필드만 선택
import requests

url = "https://api.example.com/graphql"
query = """
query($after: String){
  products(first: 50, after: $after){
    pageInfo{ hasNextPage endCursor }
    nodes{ id name price currency }
  }
}
"""
variables = {"after": None}
res = requests.post(url, json={"query": query, "variables": variables}).json()

고빈도·지연 민감한 스트림은 WebSocket을 고려(예: 시세/알림).

5.3 웹 스크래핑(페이지에서 직접 추출)

• 정적 페이지: requests + BeautifulSoup(lxml)로 빠르고 가벼움.
• 동적 페이지(JS 렌더링): Selenium/Playwright로 렌더 후 추출.
• 주의: 서비스 약관/robots.txt, 로그인/쿠키, 차단 회피(합법 범위), 구조 변화 내성.

# 정적 페이지 스크래핑 예시
import requests
from bs4 import BeautifulSoup

html = requests.get("https://example.com/list", timeout=10).text
soup = BeautifulSoup(html, "lxml")
rows = []
for card in soup.select(".item-card"):
    rows.append({
        "title": card.select_one(".title").get_text(strip=True),
        "price": card.select_one(".price").get_text(strip=True),
        "url": card.select_one("a")["href"],
    })

# 동적 페이지 스크래핑 예시 (Selenium)
from selenium import webdriver
from selenium.webdriver.common.by import By

driver = webdriver.Chrome()
driver.get("https://example.com/dynamic")
items = []
for el in driver.find_elements(By.CSS_SELECTOR, ".row"):
    items.append(el.text)
driver.quit()

---

6) 적용 가이드라인(체크리스트)

6.1 수집 전략 선택 의사결정 순서

1. 목적 정의: 어떤 의사결정/모델/리포트를 위해 무엇이 필요한가.
2. 데이터 위치: 파일? API? 웹? 복수 결합?
3. 신선도/빈도: 배치(일/주) vs 실시간/근실시간.
4. 품질·보안: 인증, 민감정보, 로그 추적성, 원천 무결성.
5. 예산/복잡도: 유지관리 가능성(스키마 변화, 페이지 구조 변경).

6.2 품질 확보 절차(최소 권장)

• 스키마 계약(필드명/타입/널 허용/코드체계) 문서화
• 검증 규칙(유효 범위, 정규식, 참조 무결성) 적용
• 중복/결측 처리: 키 기준 중복 제거, 비즈니스 규칙에 따른 보간/삭제
• 로깅/메타데이터: 수집 시각, 소스, 해시(무결성) 기록
• 재현성: 동일 입력→동일 결과(버전 고정, 컨테이너/워크플로우 정의)

---

7) 파일·API·스크래핑 비교

• 파일: 단순·저비용·배치 적합. 단, 최신성/동기화는 별도 관리 필요.
• API(REST/GraphQL/WebSocket): 공식 채널·신뢰 높음·자동화 용이. 단, Rate limit/권한/비용 고려.
• 스크래핑: API 미제공 시 대안. 구조 변경·차단·법적 이슈 리스크. 캐시/백오프/모니터링 필수.

---

8) 미니 실습 과제

1. CSV+JSON 병합: 두 소스 키 조인 후, 날짜 캐스팅·결측 처리·유효성 검증 보고서 생성.
2. REST API 수집 파이프라인: 페이징·재시도(지수 백오프)·에러 로깅·증분 수집 구현.
3. 스크래핑 견고화: CSS 셀렉터 변경에 대비해 XPATH/백업 셀렉터와 구조 검출 알림 추가.

---

9) 핵심 정리

• 데이터 수집은 목적 적합 + 품질 보장의 전략적 과정.
• 정형/반정형/비정형의 구조적 차이를 이해하고, 이에 맞는 파일·API·스크래핑 방식을 선택.
• 품질 요건(정확성·완전성·일관성·유효성·적시성·상호운용성)을 충족하도록 스키마·검증·로깅을 체계화.

---

10) 약어 해설

• API(Application Programming Interface): 애플리케이션 간 상호작용 규약
• HTTP(Hypertext Transfer Protocol): 웹 통신 프로토콜
• JSON(JavaScript Object Notation): 경량 데이터 교환 형식
• XML(eXtensible Markup Language): 확장 가능한 마크업 언어
• NLP(Natural Language Processing): 자연어 처리
• RDBMS(Relational Database Management System): 관계형 DBMS

---

오픈소스 기반 데이터 분석 4강 — 데이터 수집

---

4-1 CSV 파일 읽기

# --- pandas로 CSV 읽기: 가장 실무적인 기본형 ---
# 역할: CSV를 DataFrame으로 읽어서 표 형태로 다루기
# 주요 옵션:
# - sep: 구분자(기본 ','); 탭은 '\t'
# - header: 헤더 행 번호 또는 None
# - names: 헤더가 없을 때 컬럼명 리스트 지정
# - index_col: 인덱스 컬럼 지정(정수/이름/리스트)
# - usecols: 필요한 컬럼만 로드(메모리 절약)
# - dtype: 컬럼별 타입 강제
# - parse_dates: 날짜 컬럼 파싱(True/리스트/딕셔너리)
# - encoding: 'utf-8', 'cp949'(윈도 CSV 자주)
# - na_values: 결측치로 처리할 값 목록
# - nrows/skiprows: 일부만 읽기(샘플링/스킵)
import pandas as pd

df = pd.read_csv(
    'data.csv',
    sep=',',
    header=0,
    index_col=None,
    encoding='utf-8',
    skiprows=None,
    nrows=None,
    # names=['날짜', '체중', '골격근량', '체지방량'],    # header가 없으면 사용
    # usecols=['날짜', '체중'],                     # 필요한 열만
    # dtype={'체중':'float64'},
    # parse_dates=['날짜'],                        # 날짜 파싱
)
print(df)

# 대안 1: 표준 csv 모듈(속도↓, 의존성↓)
# [import csv]
# import csv
# with open('data.csv', encoding='utf-8') as f:
#     reader = csv.DictReader(f)
#     rows = list(reader)

# 대안 2: pyarrow/csv(대용량·고성능)
# [import pyarrow]
# import pyarrow.csv as pv
# import pyarrow.parquet as pq
# table = pv.read_csv('data.csv')
# pq.write_table(table, 'data.parquet')  # 컬럼형 포맷으로 변환

# 대안 3: numpy(수치형 배열 위주)
# [import numpy]
# import numpy as np
# arr = np.genfromtxt('data.csv', delimiter=',', names=True, dtype=None, encoding='utf-8')

---

4-2 JSON 파일 읽기

# --- 표준 json + pandas.read_json 비교 ---
# 역할: (1) 파일로부터 JSON 로드, (2) DataFrame으로 변환

import json
import pandas as pd

# 1) 파일에서 JSON 객체 로드(dict/list 등)
with open('data.json', mode='r', encoding='utf-8') as f:
    data = json.load(f)             # 역할: JSON 문자열 → 파이썬 객체
print(data)                         # 구조 확인(중첩 여부 확인)

# 2) DataFrame으로 읽기
# orient: 'records'가 가장 직관적(레코드 리스트)
# lines=True면 JSON Lines 포맷(한 줄당 한 객체)
df = pd.read_json(
    'data.json',
    orient='records',
    lines=False,
    encoding='utf-8',
)
print(df)

# 중첩 구조 평탄화(칼럼 펼치기)
# [import pandas]
# from pandas import json_normalize
# flat = pd.json_normalize(data, record_path=['매출데이터'])  # '매출데이터' 배열을 표로
# print(flat)

# 대안: HTTP로 JSON 받기
# [import requests]
# import requests
# resp = requests.get('https://example.com/api.json', timeout=10)
# resp.raise_for_status()
# data = resp.json()
# df = pd.json_normalize(data)

---

4-3 텍스트 파일 읽기 및 데이터 추출(정규식 마스킹)

# --- 로그 텍스트에서 주민등록번호 마스킹 ---
# 역할: 민감정보(패턴)를 찾아 일부만 남기고 가리기
# 주의: 실제 서비스는 더 강력한 PII(개인정보) 마스킹 정책 필요

import re

# 파일 읽기
with open('callcenter20250301.log', 'r', encoding='utf-8') as f:
    content = f.read()

# 주민등록번호 패턴(예: 6자리-7자리). 실제론 유효성 검증까지 고려 권장
pattern = re.compile(r'(\d{6})-(\d{7})')

# 그룹1(앞6자리)만 남기고 뒤 7자리는 별표로 대체
masked_content = pattern.sub(r'\1-*******', content)

# 결과 저장(확장자 일관성 권장: .log 또는 .txt 중 택1)
with open('callcenter20250301_masked.log', mode='w', encoding='utf-8') as f:
    f.write(masked_content)

print("주민등록번호 마스킹 완료. 'callcenter20250301_masked.log' 로 저장되었습니다.")

# 옵션/대안:
# - re.IGNORECASE, re.MULTILINE 등 플래그 사용: re.compile(pat, re.M|re.I)
# - 함수 기반 치환으로 부분 마스킹 강도 조절:
#   def mask(m): return f"{m.group(1)}-{'*'*len(m.group(2))}"
#   masked = pattern.sub(mask, content)
# - 대용량 파일은 스트리밍 처리 권장(라인 단위 읽기)
# - [import pathlib] Path로 경로 조작, [import io]로 버퍼링 제어

---

4-4 Open-Meteo 무료 날씨 API로 현재 온도 조회

# --- requests로 공공 API 호출 ---
# 역할: 위경도 좌표의 현재 기온을 JSON으로 받아 출력
# 주의: URL은 엔드포인트만 두고, 쿼리는 params에 분리하여 가독성과 안전성 확보
# 참고 옵션:
# - timeout: 네트워크 지연 대비
# - response.raise_for_status(): HTTP 에러 처리
# - timezone: 'Asia/Seoul' 지정 시 현지 시간대 응답

# [import requests]
import requests

BASE_URL = "https://api.open-meteo.com/v1/forecast"
params = {
    "latitude": 37.58638333,
    "longitude": 127.0203333,
    "current": "temperature_2m",      # 현재 기온 필드만 요청
    "timezone": "Asia/Seoul",
}

try:
    resp = requests.get(BASE_URL, params=params, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    temp = data['current']['temperature_2m']
    unit = data['current_units']['temperature_2m']
    print("API 응답:", data)
    print(f"서울시 종로구의 현재 온도는 : {temp}{unit} 입니다.")
except requests.exceptions.RequestException as e:
    print(f"API 호출 실패: {e}")
except (KeyError, TypeError) as e:
    print(f"응답 스키마 변경/누락 가능성: {e}")

# 대안:
# - [import httpx] (비동기/타임아웃/재시도 설정 편리)
#   import httpx
#   with httpx.Client(timeout=10) as client:
#       r = client.get(BASE_URL, params=params)
#       r.raise_for_status()
#       data = r.json()
# - pandas로 바로 normalize
#   [import pandas]
#   import pandas as pd
#   from pandas import json_normalize
#   df = json_normalize(data)

---

4-5 Selenium + lxml로 웹 스크래핑(동적 페이지)

# --- Selenium(headless Chrome)으로 동적 페이지 렌더링 + lxml로 파싱 ---
# 역할: 자바스크립트 렌더링이 필요한 페이지를 브라우저로 열어 HTML을 얻고, XPath로 요소 추출
# Colab/서버 환경에서 headless 옵션 필수적일 때가 많음

from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromeService
from selenium.webdriver.common.by import By
# [import webdriver_manager]
from webdriver_manager.chrome import ChromeDriverManager
# [import lxml]
from lxml import html
import time

chrome_options = webdriver.ChromeOptions()
chrome_options.add_argument('--headless')               # 창 없이 실행
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-dev-shm-usage')  # 공유 메모리 부족 보호
chrome_options.add_argument('--window-size=1920x1080')
# 환경에 따라 바이너리 위치 지정이 필요할 수 있음
# chrome_options.binary_location = "/usr/bin/google-chrome-stable"

# 드라이버 실행(로컬 크롬 버전에 맞는 드라이버 자동 설치)
driver = webdriver.Chrome(
    service=ChromeService(ChromeDriverManager().install()),
    options=chrome_options
)

url = 'https://curiousweek.tistory.com/357'
driver.get(url)

# 명시적 대기 권장(렌더 완료까지)
# [import selenium] WebDriverWait/EC 사용 예:
# from selenium.webdriver.support.ui import WebDriverWait
# from selenium.webdriver.support import expected_conditions as EC
# WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.TAG_NAME, 'title')))

time.sleep(2)  # 단순 대기(학습용)

# HTML 파싱 후 제목 추출
page_source = driver.page_source
tree = html.fromstring(page_source)
title_text = tree.xpath('//title/text()')
print(title_text)

driver.quit()

# 대안:
# - 정적 페이지: requests + BeautifulSoup가 더 간단/빠름
#   [import requests, bs4]
#   import requests
#   from bs4 import BeautifulSoup
#   r = requests.get(url, timeout=10); r.raise_for_status()
#   soup = BeautifulSoup(r.text, 'html.parser')
#   print(soup.title.get_text(strip=True))
# - Playwright(안정적·강력): [import playwright]
# - parsel(강력 XPath/CSS 선택자): [import parsel]

---

4-6 공공데이터 API(경기데이터드림 예시) 호출

# --- 공개 API 호출: 페이지네이션/에러 처리 포함 예시 ---
# 역할: 경기도 지역화폐 가맹점 현황(예시 엔드포인트) 호출 후 JSON 확인
# 주의:
# - 실제 서비스키(KEY)는 노출 금지. 환경변수/비밀관리 사용 권장.
# - 응답 스키마(키 이름)는 기관별로 상이 → print로 구조 먼저 확인 후 파싱

# [import requests, os]
import requests
import os

url = 'https://openapi.gg.go.kr/RegionMnyFacltStus'
api_key = os.getenv('GG_API_KEY', 'c79d')  # 데모용 기본값(실습 시 자신의 키로 대체)

params = {
    'KEY': api_key,     # 인증키
    'Type': 'json',     # 응답 타입
    'pIndex': 1,        # 페이지 번호(1부터)
    'pSize': 10         # 페이지 크기
    # 추가 검색 파라미터는 기관 문서 참조(예: SIGUN_NM=수원시)
}

resp = requests.get(url, params=params, timeout=10)
resp.raise_for_status()
payload = resp.json()
print(payload)  # 구조 확인이 먼저

# --- 필요 데이터만 DataFrame으로 변환 (가변 스키마 고려) ---
# 기관별 JSON 구조가 달라서 KeyError 방지용으로 방어적 접근
# [import pandas]
import pandas as pd

root_key = 'RegionMnyFacltStus'
if isinstance(payload, dict) and root_key in payload:
    # 보통 [0]에 'head', [1]에 'row'가 들어있는 패턴이 있음(기관별 상이)
    blocks = payload[root_key]
    rows = None
    for blk in blocks:
        if 'row' in blk:
            rows = blk['row']
            break
    if rows:
        df = pd.DataFrame(rows)
        print(df.head())
    else:
        print('row 블록을 찾지 못했습니다. 응답 구조를 확인하세요.')
else:
    print('예상한 루트 키가 없습니다. 응답 구조를 확인하세요.')

# 페이지네이션(다건 수집) 예시
# for page in range(1, 6):
#     params['pIndex'] = page
#     r = requests.get(url, params=params, timeout=10)
#     r.raise_for_status()
#     data = r.json()
#     # 누적 로직 작성(append/concat) → 대량이면 리스트로 모아 마지막에 concat 권장

# 대안:
# - httpx(Client, 재시도/타임아웃 관리 용이) [import httpx]
# - pydantic으로 스키마 검증 [import pydantic]
# - 요청 서명/헤더 필요한 API는 requests.get(url, headers=...)로 확장