이 글에서는 PySpark와 Bright Data를 사용하여 대규모 웹 스크래핑 작업을 실행하는 방법을 다룹니다. 수십만 개의 상품 페이지를 스크래핑하거나, 수백 개 사이트의 가격을 모니터링하거나, 수백만 개의 페이지에서 훈련 데이터셋을 구축해야 하는 경우, 단일 머신 스크립트만으로는 이를 달성할 수 없습니다.
이 글에서 소개하는 패턴을 통해 요청량이 증가해도 파이프라인의 안정성을 유지하면서 스크래핑 작업을 클러스터 전체에 분산하는 방법을 확인할 수 있습니다.
이 글을 마치면 다음을 수행할 수 있게 됩니다:
- PySpark를 사용하여 대규모 URL 목록을 분산 데이터셋으로 처리하는 방법
- 파티션 수준에서 스크래핑 워크로드를 효율적으로 실행하는 방법
- 전체 작업을 재시작하지 않고도 재시도 및 오류를 처리할 수 있는 워커 설계하기
- 요청량이 증가함에 따라 프록시 라우팅 및 네트워크 안정성을 처리하는 방법
웹 스크래핑이 분산 문제로 변할 때
대부분의 스크래핑 프로젝트는 동일한 방식으로 시작됩니다. 개발자가 스크립트를 작성하고, URL 목록을 읽어들이며, 요청을 전송하고, 결과를 저장하는 식입니다.
워크로드가 확장되면 문제점이 드러납니다. 예전에는 몇 분이면 끝났던 작업이 몇 시간이나 걸리기 시작합니다. 수천 개의 페이지를 처리한 후 몇 개의 요청 실패만으로도 실행이 중단될 수 있으며, 데이터 가져오기와 파싱을 처리하면서 동일한 스크립트 내에서 재시도를 관리하는 일은 금세 난장판이 됩니다. 저는 팀들이 이러한 단일 파일 스크래퍼를 몇 달 동안 유지 관리하며, 하나씩 예외적인 상황을 수정해 나가는 모습을 보아왔습니다. 하지만 진짜 문제는 아키텍처가 더 이상 문제에 맞지 않는다는 점입니다.
스레딩을 사용하더라도 한 대의 머신에서 수십만 개의 페이지를 스크래핑하는 데는 비현실적으로 긴 시간이 소요됩니다. 대규모 환경에서는 여러 워커에 걸쳐 작업을 실행해야 하며, 일부 요청이 실패하더라도 시스템은 계속 가동되어야 합니다. 앞으로 나아갈 방향은 URL 목록을 순차적인 큐로 생각하지 말고, 분산 처리할 수 있는 데이터셋으로 취급하기 시작하는 것입니다.
PySpark가 여기에 적합한 이유
PySpark는 데이터셋을 파티션으로 분할하여 머신 클러스터 전반에서 병렬로 처리한다는 개념을 기반으로 구축되었습니다. 이 모델은 웹 스크래핑에 직접적으로 적용됩니다. 각 URL은 작업 단위이며, 파티션은 URL을 배치로 묶고, 익스큐터는 해당 배치를 독립적으로 처리합니다.
Celery나 자체 제작한 멀티프로세싱 설정을 통해 큐를 관리하는 대신, Spark는 별도의 구축 과정 없이도 내결함성과 스케줄링 기능을 제공합니다. 작업이 실패하면 Spark가 재스케줄링하고, 노드가 다운되면 작업이 재할당됩니다. 작업 내부에 적절한 재시도 로직을 작성해야 하지만, 오케스트레이션 계층은 자동으로 처리됩니다.
패턴 1: 분산 데이터셋으로서의 URL
모든 분산 스크래핑 파이프라인의 기초는 URL 목록을 어떻게 로드하느냐에 달려 있습니다. PySpark를 사용하면 URL이 DataFrame에 담겨지고, Spark는 이를 작업자(worker)들 간에 자동으로 분배합니다. 각 파티션은 데이터의 일부를 담고 있으며, Spark는 사용 가능한 실행자(executor)들에게 해당 파티션을 할당합니다.
기본 설정은 다음과 같습니다:
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("distributed_scraper").getOrCreate()
urls = [
("https://example.com/page1",),
("https://example.com/page2",),
("https://example.com/page3",)
]
df = spark.createDataFrame(urls, ["url"])실제 운영 환경에서는 URL 목록을 하드코딩하지 않고 파일, 데이터베이스 테이블 또는 오브젝트 스토리지에서 불러와야 합니다. 크롤링 우선순위나 마지막 가져온 타임스탬프와 같은 메타데이터를 추가하기 시작하면 스키마도 중요해집니다.
파티션 수는 가장 먼저 결정해야 할 튜닝 사항입니다. 파티션 수가 너무 적으면 워커들이 느린 요청을 기다리며 유휴 상태가 되고, 너무 많으면 Spark가 실제 데이터 가져오기보다 스케줄링 오버헤드에 불균형적으로 많은 시간을 소비하게 됩니다.
스크래핑 워크로드의 합리적인 시작점은 실행기 코어당 2~4개의 파티션이며, 이후 작업 로그를 바탕으로 조정하면 됩니다. 실행기가 1초 이내에 파티션을 완료하거나 지속적으로 10분 이상 소요된다면 파티션 크기를 조정해야 합니다.
패턴 2: 파티션 수준에서 요청 실행
가장 먼저 시도해 볼 수 있는 방법은 DataFrame 내 각 URL에 행(row) 수준 변환을 적용하는 것입니다. 이 접근 방식은 작동하지만 웹 스크래핑에는 적합하지 않습니다. 각 요청은 별도의 함수 호출을 유발하므로, 주의하지 않으면 URL마다 새로운 연결이 생성됩니다. 수백만 행에 이르면 오버헤드가 급격히 증가합니다.
올바른 접근 방식은 mapPartitions()를 사용하는 것입니다 . 이 함수는 한 번에 한 행씩 처리하는 대신, 전체 파티션을 반복자(iterator) 형태로 함수에 전달합니다. HTTP 세션을 한 번 생성한 후 해당 파티션 내의 모든 요청에 재사용할 수 있습니다. 특히 HTTP keep-alive를 지원하는 서버의 경우, 각 URL마다 새로운 TCP 연결을 설정하는 것보다 장시간 실행되는 세션을 통한 연결 풀링이 훨씬 더 빠릅니다.
from pyspark.sql import SparkSession
import requests
spark = SparkSession.builder.appName("distributed_scraper").getOrCreate()
urls = [
("https://example.com/page1",),
("https://example.com/page2",),
("https://example.com/page3",)
]
df = spark.createDataFrame(urls, ["url"])
def scrape_partition(rows):
session = requests.Session()
for row in rows:
url = row["url"]
try:
response = session.get(url, timeout=30)
yield {
"url": url,
"status_code": response.status_code,
"html": response.text
}
except Exception:
yield {
"url": url,
"status_code": None,
"html": None
}
results = df.rdd.mapPartitions(scrape_partition)요청이 실패하면 예외를 발생시키는 대신 null 필드가 포함된 레코드를 반환합니다. 이는 의도된 접근 방식입니다. 예외가 전파되면 전체 파티션 작업이 중단되어 실패 전까지 수행된 모든 작업이 손실됩니다. null 레코드를 반환하면 파티션이 계속 실행되며, 나중에 실패한 URL을 식별하고 재시도할 수 있는 깔끔한 방법을 제공합니다.
초기 단계에서 수행할 가치가 있는 작업 중 하나는 Spark가 RDD에서 스키마를 추론하도록 두는 대신 StructType을 사용하여 명시적인 출력 스키마를 정의하는 것입니다. 스키마 추론은 데이터의 전체 스캔을 필요로 하여 비용이 많이 들며, 응답 내용이 예기치 않게 비어 있을 때 가끔 예상치 못한 결과를 초래할 수 있습니다.
패턴 3: 장시간 실행을 처리할 수 있는 워커 설계
백만 개의 페이지를 스크래핑하는 작업은 몇 시간 동안 실행됩니다. 장시간 실행 중에는 연결 재설정, DNS 타임아웃, 속도 제한 서버에서 발생하는 429 오류, 응답 도중 서버가 가끔 연결을 끊는 현상 등을 목격하게 될 것입니다. 이 중 어느 것도 코드의 버그가 아닙니다. 대규모로 HTTP 요청을 수행할 때 발생하는 정상적인 현상일 뿐입니다.
이러한 모든 문제를 처리하기에는 파티션 함수가 가장 적합한 장소입니다. 재시도 로직, 백오프 지연, 타임아웃 설정, 오류 기록 등은 모두 이곳에 구현되어야 합니다. 모든 것을 단일 파티션 함수에 포함하면 나머지 Spark 파이프라인을 깔끔하게 유지할 수 있으며, 워커의 동작을 독립적으로 테스트할 수 있습니다.
import requests
import time
def scrape_partition(rows):
session = requests.Session()
for row in rows:
url = row["url"]
attempts = 0
success = False
while attempts < 3 and not success:
try:
response = session.get(url, timeout=30)
yield {
"url": url,
"status_code": response.status_code,
"html": response.text
}
success = True
except Exception as e:
attempts += 1
time.sleep(2 ** attempts) # 지수적 백오프
if not success:
yield {
"url": url,
"status_code": None,
"html": None
}여기서 몇 가지 주의할 점이 있습니다. 재시도 지연 시간은 고정된 대기 시간 대신 지수적 백오프를 사용합니다. 가끔 발생하는 네트워크 일시적 장애에는 2초의 고정 지연 시간도 괜찮지만, 지속적으로 트래픽 제한이 걸린 서버에 접속할 경우 작업자(worker)의 속도가 상당히 느려집니다. 또한, null 레코드를 반환하기 전에 예외 유형을 기록해 두세요. 연결 시간 초과와 403 Forbidden의 차이는 업스트림에서 무슨 일이 일어나고 있는지에 대해 매우 다른 정보를 제공합니다.
프로덕션 환경에서의 작업 모니터링
작업이 수 시간에 걸쳐 수백만 개의 URL을 처리할 때는 실행 중에 어떤 일이 일어나고 있는지 파악할 수 있어야 합니다. 최소한 각 파티션에서 다음 메트릭을 추적해야 합니다:
def scrape_partition(rows):
session = requests.Session()
partition_stats = {
"urls_attempted": 0,
"urls_succeeded": 0,
"urls_failed": 0,
"status_codes": {}
}
for row in rows:
partition_stats["urls_attempted"] += 1
url = row["url"]
attempts = 0
success = False
while attempts < 3 and not success:
try:
response = session.get(url, timeout=30)
partition_stats["urls_succeeded"] += 1
code = response.status_code
partition_stats["status_codes"][code] =
partition_stats["status_codes"].get(code, 0) + 1
yield {
"url": url,
"status_code": response.status_code,
"html": response.text
}
success = True
except Exception as e:
attempts += 1
time.sleep(2 ** attempts)
if not success:
partition_stats["urls_failed"] += 1
yield {
"url": url,
"status_code": None,
"html": None
}
# 파티션 완료 시 통계 기록
print(f"Partition stats: {partition_stats}")작업이 실행되는 동안 Spark UI에서 작업 완료율을 확인하세요. 작업이 매우 다른 속도로 완료된다면 파티션이 불균형한 것입니다. 로그에 403 또는 429 오류가 지속적으로 나타난다면 프록시 로테이션을 조정하거나 요청 지연 시간을 추가해야 합니다. 목표는 작업이 실패한 지 6시간 후에 문제를 발견하는 것이 아니라, 작업이 아직 실행 중일 때 문제를 포착하는 것입니다.
워커의 결과 쓰기 (프로덕션 패턴)
1시간 이상 실행되는 작업의 경우, 드라이버 프로세스가 실행 도중 종료되는 실패 모드는 재시도 로직으로도 방지할 수 없습니다. Spark는 개별 태스크가 실패하면 재스케줄링하지만, 드라이버가 다운되면 작업 전체가 손실됩니다.
해결책은 모든 데이터를 드라이버로 다시 전송하고 작업이 완료될 때까지 결과를 메모리에 보관하는 대신, 각 파티션이 완료될 때마다 결과를 영구 저장소에 기록하는 것입니다. foreachPartition()을 사용하면 각 파티션을 처리하고, 데이터가 드라이버를 거치지 않고 워커에서 직접 출력을 기록할 수 있습니다:
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
import requests, time, uuid
spark = SparkSession.builder.appName("distributed_scraper").getOrCreate()
spark.sparkContext.setCheckpointDir("s3://your-bucket/checkpoints/")
schema = StructType([
StructField("url", StringType(), True),
StructField("status_code", IntegerType(), True),
StructField("html", StringType(), True)
])
def scrape_and_write(rows):
session = requests.Session()
results = []
for row in rows:
url = row["url"]
attempts = 0
success = False
while attempts < 3 and not success:
try:
response = session.get(url, timeout=30)
results.append((url, response.status_code, response.text))
success = True
except Exception as e:
시도 횟수 += 1
time.sleep(2 ** 시도 횟수)
if 성공:
결과.append((url, None, None))
# 이 파티션의 결과를 워커에서 직접 작성
partition_id = str(uuid.uuid4())
spark.createDataFrame(results, schema).write.mode("append").parquet(
f"s3://your-bucket/scrape-results/batch={partition_id}"
)
df.rdd.foreachPartition(scrape_and_write)각 워커는 자체 출력 파일을 독립적으로 작성합니다. 드라이버가 작업 도중에 종료되더라도 완료된 파티션은 이미 스토리지에 저장되어 있으므로, 진행 중인 파티션만 다시 실행하면 됩니다. 스크랩된 데이터에 대해 다운스트림 Spark 변환이 수행되는 작업의 경우, rdd.checkpoint()를 사용하는 것이 더 가벼운 대안입니다. 이 메서드는 변환이 실행되기 전에 RDD를 체크포인트 디렉터리에 구체화하여, 후속 단계에서 오류가 발생하더라도 Spark가 전체 스크랩 단계를 다시 실행하지 않도록 방지합니다.
패턴 4: 프록시 네트워크를 통한 요청 라우팅
여러 워커를 병렬로 실행하면 처리량이 증가하지만, 대상 서버는 클러스터의 IP 범위에서 쏟아지는 요청을 감지하게 됩니다. 대부분의 사이트는 단일 IP 범위에서 집중적으로 유입되는 이러한 트래픽 패턴을 차단하거나 속도 제한을 적용하도록 구성되어 있습니다. 주거용 프록시 네트워크를 통해 요청을 라우팅하면 트래픽이 여러 IP 주소로 분산되어, 차단 없이 워커를 계속 실행할 수 있습니다.
파티션 함수 내에서 세션당 한 번만 프록시를 구성하면, 해당 세션에서 발생하는 모든 요청이 자동으로 네트워크를 통해 라우팅됩니다:
import requests
BRIGHTDATA_PROXY = (
"http://brd-customer-<customer_id>-zone-<zone_name>:"
"<zone_password>@brd.superproxy.io:33335"
)
def scrape_partition(rows):
session = requests.Session()
session.proxies = {
"http": BRIGHTDATA_PROXY,
"https": BRIGHTDATA_PROXY
}
for row in rows:
url = row["url"]
try:
response = session.get(url, timeout=30)
yield {
"url": url,
"status_code": response.status_code,
"html": response.text
}
except Exception as e:
yield {
"url": url,
"status_code": None,
"html": None
}Bright Data 존 구성에 따라 트래픽이 중간 인증서 계층을 통과하기 때문에 요청 시 SSL 검증 오류가 발생할 수 있습니다. 빠른 해결책은 verify=False를 전달하고 진행하는 것이지만, 이 방법은 인증서 검증을 완전히 비활성화하므로 워커가 프록시와 대상 간의 연결이 손상되었는지 더 이상 감지할 수 없게 됩니다.
올바른 해결 방법은 Bright Data의 CA 인증서를 다운로드하여 verify='/path/to/brightdata-ca.crt'를 통해 전달하는 것으로, 이렇게 하면 전체 인증 검증이 유지됩니다. 또한 주목할 점은, 예제의 프록시 URL은 실제 운영 환경에서는 환경 변수나 시크릿 관리자를 통해 가져와야 한다는 것입니다. 분산 환경에서는 이러한 자격 증명이 직렬화되어 모든 워커 노드로 전송되므로, 정보 유출 시 단일 머신에서보다 더 큰 피해를 입힐 수 있습니다.
자바스크립트로 렌더링된 콘텐츠를 제공하는 대상의 경우, 표준 프록시를 통한 라우팅만으로는 충분하지 않습니다. Bright Data의 Scraping Browser는 자바스크립트 실행, CAPTCHA 해결 및 브라우저 지문 인식 기능을 처리하며, Playwright 및 Puppeteer와 통합됩니다. 파티션 함수 구조는 동일하게 유지되며, 요청 세션을 Scraping Browser 엔드포인트를 가리키는 Playwright 브라우저 인스턴스로 교체하기만 하면 됩니다.
일반적인 문제 해결
실제 운영 환경에서 몇 가지 문제가 지속적으로 발생합니다. 파티션 작업이 반복적으로 타임아웃된다면 먼저 파티션 크기를 확인하십시오. 10,000개 이상의 URL을 포함하는 파티션은 요청 속도가 느릴 경우 Spark의 기본 타임아웃을 초과하게 됩니다. 더 작은 배치로 재분할하거나 spark.task.maxFailures 및 spark.network.timeout 값을 늘리십시오.
프록시를 사용 중임에도 429 오류가 발생한다면, 여러 워커가 동시에 동일한 도메인에 접속하고 있다는 의미입니다. 요청 사이에 무작위 지터(jitter)를 추가하세요:
import random
import time
def scrape_partition(rows):
session = requests.Session()
for row in rows:
time.sleep(random.uniform(1, 3))
# ... 나머지 스크래핑 로직익스큐터에서 메모리 오류가 발생한다면, 일반적으로 HTML을 모두 축적한 후에 작성하고 있다는 의미입니다. 결과를 더 자주 작성하거나, 추출된 필드만 필요한 경우 파티션 함수 내에서 HTML을 파싱하고 버리십시오.
파티션별 완료 속도가 현저히 다른 경우, 데이터 분배가 불균형함을 나타냅니다. 느린 도메인이 여러 워커에 분산되도록 파티션 수를 늘려 재분할하십시오.
마치며
이러한 패턴은 대규모 환경에서도 견고하게 작동하는 기반을 제공합니다: URL 목록을 분산하고, 파티션 수준에서 요청을 실행하며, 장시간 실행에도 견딜 수 있는 워커를 구축하고, 트래픽이 증가해도 차단되지 않는 프록시 네트워크를 통해 트래픽을 라우팅하십시오.
실전 환경의 작업에는 명시적인 스키마, 체크포인트링, 적절한 시크릿 관리가 필요하지만, 규모와 관계없이 구조적인 결정 사항은 동일합니다. 네트워크 및 인프라 측면에서는 Bright Data가 직접 구축하고 유지 관리해야 할 대부분의 요소를 처리해 줍니다.
