파이썬 웹 스크레이핑: 하루 만에 수백만 건 데이터 수집하기

데이터 분석, 시장 동향 파악, 특정 분야의 발전 양상 이해 등 다양한 목적으로 대량의 데이터를 수집해야 할 때가 있습니다. 수동으로 데이터를 모으는 것은 시간이 많이 들고 비효율적이며, 데이터의 정확성과 완전성을 보장하기 어렵습니다. 이러한 대규모 데이터 수집 작업을 빠르고 효율적으로 수행할 수 있는 방법은 없을까요?

파이썬은 풍부한 서드파티 라이브러리와 도구를 제공하며, 그중 웹 스크레이핑 라이브러리는 특히 강력합니다. 본 문서에서는 파이썬 웹 스크레이핑을 활용하여 하루 만에 수백만 건의 데이터를 수집하는 방법에 대해 설명합니다.

데이터 소스 정의

데이터 수집을 시작하기 전에 데이터 소스를 명확히 정의해야 합니다. 데이터 소스는 웹사이트, API, 데이터베이스 등 다양할 수 있습니다. 여기서는 웹사이트를 예로 들어 설명합니다.

예를 들어, 특정 전자상거래 웹사이트에서 상품명, 가격, 판매량, 리뷰 등 상품 정보를 수집한다고 가정해 봅시다. 먼저 해당 웹사이트의 URL과 페이지 구조를 파악해야 합니다. 웹 페이지 소스 코드를 분석하여 상품 정보가 HTML 태그 내에 저장되어 있으며, 각 상품이 고유한 URL을 가지고 있음을 확인할 수 있습니다. 따라서 HTML 태그와 URL 링크를 파싱하여 상품 정보를 수집할 수 있습니다.

기본 스크레이퍼 프로그램 구현

데이터 소스를 정의했다면 이제 스크레이퍼 프로그램을 작성할 차례입니다. 스크레이퍼 프로그램은 주로 다음 단계로 구성됩니다.

  1. HTTP 요청 전송: 파이썬 requests 라이브러리를 사용하여 HTTP 요청을 보내고 웹 페이지의 HTML 소스를 가져옵니다.

    import requests

target_url = 'https://www.example.com/products'
try:
    page_response = requests.get(target_url, timeout=10)
    page_response.raise_for_status() # HTTP 오류 발생 시 예외 발생
    page_html_content = page_response.text
except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")
    page_html_content = None

  2. HTML 콘텐츠 파싱: 파이썬 BeautifulSoup 라이브러리를 활용하여 HTML 태그를 파싱하고 필요한 정보를 추출합니다. 실제 반환되는 콘텐츠 구조에 따라 분석 및 수정이 필요합니다.

    from bs4 import BeautifulSoup

if page_html_content:
    soup_parser = BeautifulSoup(page_html_content, 'html.parser')
    product_listings = soup_parser.find_all('div', class_='product-card')

    extracted_data = []
    for product in product_listings:
        item_name = product.find('h2', class_='product-name').text.strip() if product.find('h2', class_='product-name') else 'N/A'
        item_price = product.find('span', class_='price-value').text.strip() if product.find('span', class_='price-value') else 'N/A'
        item_sales_count = product.find('div', class_='sales-count').text.strip() if product.find('div', class_='sales-count') else 'N/A'
        item_rating_score = product.find('div', class_='star-rating').text.strip() if product.find('div', class_='star-rating') else 'N/A'

        extracted_data.append({
            'name': item_name,
            'price': item_price,
            'sales': item_sales_count,
            'rating': item_rating_score
        })
    # extracted_data를 활용하여 추가 작업 수행 (예: 데이터베이스 저장, 파일 저장)

  3. URL 링크 순회: 페이지네이션이 있는 경우, 여러 페이지에 걸쳐 상품 정보를 수집하기 위해 파이썬의 urllib.parse 또는 직접 URL 조작을 통해 URL 링크를 순회합니다.

    import requests
from bs4 import BeautifulSoup

base_listing_url = 'https://www.example.com/category?page='
total_pages_to_crawl = 50 # 예시로 50페이지까지
all_collected_items = []

for page_num in range(1, total_pages_to_crawl + 1):
    current_page_url = base_listing_url + str(page_num)
    print(f"Crawling page: {current_page_url}")
    try:
        page_response = requests.get(current_page_url, timeout=10)
        page_response.raise_for_status()
        soup_obj = BeautifulSoup(page_response.text, 'html.parser')
        items_on_page = soup_obj.find_all('div', class_='product-summary')

        for item_elem in items_on_page:
            product_title = item_elem.find('a', class_='item-title').text.strip()
            product_price_str = item_elem.find('span', class_='display-price').text.strip()
            # 추가 정보 추출 및 저장
            all_collected_items.append({'title': product_title, 'price': product_price_str})

    except requests.exceptions.RequestException as e:
        print(f"Error crawling {current_page_url}: {e}")
        continue # 다음 페이지로 이동

# all_collected_items 리스트에 수집된 데이터가 저장됩니다.

  4. 데이터 저장: 파이썬의 csv 라이브러리를 사용하여 수집된 데이터를 CSV 파일에 저장합니다.

    import csv

# 위에서 추출된 all_collected_items 리스트를 사용한다고 가정
output_filename = 'product_data_export.csv'
if all_collected_items:
    # 헤더 추출 (첫 번째 항목의 키를 사용)
    field_names = all_collected_items[0].keys()

    with open(output_filename, 'w', newline='', encoding='utf-8') as csv_file:
        writer_obj = csv.DictWriter(csv_file, fieldnames=field_names)
        writer_obj.writeheader() # 헤더 쓰기
        writer_obj.writerows(all_collected_items) # 모든 데이터 쓰기
    print(f"Data successfully saved to {output_filename}")
else:
    print("No data to save.")

스크레이퍼 효율성 향상

수백만 건의 데이터를 수집해야 할 경우, 단일 스크레이퍼 프로그램으로는 요구 사항을 충족하기 어려울 수 있습니다. 스크레이퍼의 효율성을 높이기 위해 다음과 같은 방법을 사용할 수 있습니다.

  1. 단일 머신 멀티스레딩: 멀티스레딩을 사용하면 여러 요청을 동시에 처리하여 스크레이퍼의 효율성을 높일 수 있습니다. 파이썬의 threading 라이브러리로 멀티스레딩을 구현할 수 있습니다.

    import threading
import requests
from bs4 import BeautifulSoup
import time

def fetch_and_parse_page(page_url, result_list):
    """특정 URL에서 데이터를 가져와 파싱하고 결과 리스트에 추가합니다."""
    try:
        res = requests.get(page_url, timeout=5)
        res.raise_for_status()
        parsed_soup = BeautifulSoup(res.text, 'html.parser')
        page_products = parsed_soup.find_all('div', class_='item-card')

        for product_entry in page_products:
            title_elem = product_entry.find('h3', class_='item-title')
            price_elem = product_entry.find('span', class_='item-price')
            
            title = title_elem.text.strip() if title_elem else 'Unknown Title'
            price = price_elem.text.strip() if price_elem else 'Unknown Price'
            
            result_list.append({'product_title': title, 'product_price': price, 'source_url': page_url})
        print(f"Finished crawling: {page_url}")
    except requests.exceptions.RequestException as e:
        print(f"Error fetching {page_url}: {e}")
    except Exception as e:
        print(f"Parsing error on {page_url}: {e}")

base_product_list_url = 'https://www.example.com/search?q=query&p='
total_pages_threaded = 20 # 예시로 20페이지
threaded_results = []
active_threads = []

start_time = time.time()

for p_num in range(1, total_pages_threaded + 1):
    target_page_url = base_product_list_url + str(p_num)
    thread_instance = threading.Thread(target=fetch_and_parse_page, args=(target_page_url, threaded_results))
    active_threads.append(thread_instance)
    thread_instance.start()

for t in active_threads:
    t.join() # 모든 스레드가 완료될 때까지 대기

end_time = time.time()
print(f"Total items collected: {len(threaded_results)}")
print(f"Time taken: {end_time - start_time:.2f} seconds")

# threaded_results 리스트에 모든 수집 데이터가 저장됩니다.

  2. 분산 스크레이퍼: 여러 스크레이퍼 프로그램을 동시에 실행하여 다른 웹 페이지를 크롤링함으로써 효율성을 높일 수 있습니다. 파이썬의 Scrapy 프레임워크는 분산 스크레이핑을 구현할 수 있습니다.

    Scrapy-Redis 설정 단계:

    1. 분산 프레임워크(Scrapy-Redis 또는 Scrapy-RabbitMQ)를 설치합니다.
    2. Scrapy-Redis 또는 Scrapy-RabbitMQ 연결 정보(예: Redis 서버 주소, 포트, 비밀번호 등)를 구성합니다.
    3. Scrapy 프로젝트의 settings.py 파일을 수정하여 다음 설정을 추가합니다. 
      # settings.py 예시
# Scrapy-Redis 관련 설정
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
SCHEDULER_PERSIST = True
SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderPriorityQueue" # 또는 SpiderQueue, SpiderStack

# Redis 연결 정보
REDIS_HOST = 'your_redis_server_ip'
REDIS_PORT = 6379
REDIS_PASSWORD = 'your_redis_password_if_any' # 필요 없으면 주석 처리 또는 None
REDIS_DB = 0

# 기타 Scrapy 설정
USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
ROBOTSTXT_OBEY = True # 웹사이트의 robots.txt 규칙을 준수
DOWNLOAD_DELAY = 0.5 # 요청 간 지연 시간 (초)
CONCURRENT_REQUESTS = 16 # 동시 요청 수
    4. 스파이더에서 Redis 또는 RabbitMQ URL을 지정하여 작업 분배를 구현합니다.
    5. Redis 또는 RabbitMQ 서비스를 시작합니다.
    6. 다음 명령을 사용하여 여러 스크레이퍼 노드를 시작합니다. 
      scrapy crawl your_spider_name -s JOBDIR=crawls/node_1
scrapy crawl your_spider_name -s JOBDIR=crawls/node_2
# 필요에 따라 더 많은 노드 실행
      여기서 your_spider_name은 스파이더 이름이며, -s JOBDIR=crawls/node_1은 중단된 지점부터 계속 크롤링하는 기능을 활성화하는 명령입니다.
    7. 별도의 터미널에서 다음 명령을 사용하여 초기 URL을 Redis 큐에 추가합니다. 
      # Redis 큐에 시작 URL을 추가하는 스크립트를 실행하거나
# scrapy-redis-cli를 사용합니다 (Scrapy-Redis 설치 시 포함)
# 예시: scrapy-redis-cli queue your_spider_name:start_urls < url_list.txt
# 또는 Python 스크립트로 직접 Redis에 추가

    Scrapy를 사용한 분산 스크레이핑 예시:

    1. Scrapy 프로젝트를 생성하고 위에서 언급된 설정으로 settings.py를 구성합니다.
    2. 스파이더 파일에서 크롤링할 URL 큐를 정의합니다. 
      import scrapy
from scrapy_redis.spiders import RedisSpider

class ProductCrawlerSpider(RedisSpider):
    name = 'product_crawler'
    redis_key = 'product_crawler:start_urls' # Redis에서 시작 URL을 가져올 키

    def parse(self, response):
        """
        웹 페이지 응답을 파싱하여 상품 데이터를 추출합니다.
        """
        product_cards = response.xpath('//div[contains(@class, "product-item")]')
        for card in product_cards:
            item_data = {}
            item_data['product_id'] = card.xpath('.//span[@class="id-code"]/text()').get()
            item_data['name'] = card.xpath('.//h3[@class="product-title"]/a/text()').get()
            item_data['price'] = card.xpath('.//div[@class="current-price"]/text()').get()
            item_data['reviews'] = card.xpath('.//span[@class="review-count"]/text()').get()
            
            # 상세 페이지 링크가 있다면 follow
            detail_page_link = card.xpath('.//h3[@class="product-title"]/a/@href').get()
            if detail_page_link:
                yield response.follow(detail_page_link, callback=self.parse_detail, meta={'item': item_data})
            else:
                yield item_data # 상세 페이지가 없는 경우 바로 반환

    def parse_detail(self, response):
        """
        상품 상세 페이지를 파싱하여 추가 데이터를 추출합니다.
        """
        item = response.meta['item']
        # 상세 페이지에서 추가 정보 추출
        item['description'] = response.xpath('//div[@class="product-description"]/text()').get()
        item['brand'] = response.xpath('//span[@class="product-brand"]/text()').get()
        yield item
      여기서 RedisSpider를 상속받았고, redis_key를 product_crawler:start_urls로 설정하여 Redis에서 시작 URL을 가져오도록 지정했습니다. parse 메서드에서는 XPath를 사용하여 필요한 정보를 추출하고, 상세 페이지로 이동하기 위한 response.follow도 사용했습니다. parse_detail 메서드는 상세 페이지에서 추가 정보를 추출합니다.
    3. Redis 서비스를 시작하고 크롤링할 URL을 큐에 추가합니다. 
      import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

# 크롤링할 시작 URL 목록을 Redis 큐에 추가
for i in range(1, 101): # 예시로 100페이지
    start_url_to_add = 'https://www.example.com/category/all?page=' + str(i)
    redis_client.lpush('product_crawler:start_urls', start_url_to_add)
print("Initial URLs added to Redis queue.")
      여기서는 Redis의 파이썬 클라이언트 라이브러리 redis를 사용하여 시작 URL을 product_crawler:start_urls 큐에 추가합니다.
    4. settings.py에 데이터 저장 설정을 추가합니다. 
      # settings.py 예시 (이전 내용에 추가)
FEED_FORMAT = 'jsonlines' # CSV, JSON, XML 등 다양한 형식 가능
FEED_URI = 'collected_products_%(time)s.jsonl' # 파일명에 타임스탬프 추가
FEED_EXPORT_ENCODING = 'utf-8'
      여기서는 Scrapy 자체의 JSON Lines 출력기를 사용하여 데이터를 collected_products_YYYY-MM-DDTHH-MM-SS.jsonl 파일에 저장하도록 설정했습니다. (CSV도 가능하지만 JSON Lines가 대규모 데이터에 더 유연합니다)
    5. 여러 스크레이퍼 노드를 시작합니다. 
      # 터미널 1
scrapy crawl product_crawler -s JOBDIR=crawls/product_crawler-node1

# 터미널 2
scrapy crawl product_crawler -s JOBDIR=crawls/product_crawler-node2
# 필요한 만큼 더 많은 터미널에서 실행
      여기서는 두 개의 스크레이퍼 노드를 시작했으며, -s JOBDIR 매개변수를 사용하여 각 노드가 중단 지점부터 크롤링을 재개할 수 있도록 합니다.
    6. (선택 사항) 크롤링 진행 상황 모니터링 및 Redis 큐 확인: Redis CLI 또는 redis-py를 사용하여 product_crawler:start_urls 큐의 길이 및 처리된 항목 수를 확인할 수 있습니다. 
      # Redis CLI에서
LLEN product_crawler:start_urls

    기술 요약

    위에서 설명한 방법들을 통해 대량의 데이터를 빠르고 효율적으로 수집할 수 있습니다. 물론 웹 스크레이퍼 프로그램을 작성할 때는 웹사이트의 robots.txt 규칙을 준수하고, IP 차단을 방지하는 등의 고려 사항에 주의해야 합니다. 또한 수집된 데이터를 클리닝, 분석 및 시각화하여 더욱 가치 있는 정보를 얻을 수 있습니다.

태그: python Web Scraping Scrapy Distributed Scraping requests

7월 25일 04:23에 게시됨

인기 태그

©2026 괴물 클럽