PySpark 대규모 텍스트·정규식 처리 — 수십억 로그를 파싱하기

from pyspark.sql import functions as F
 
# ❌ Python UDF (느림)
import re
@F.udf("string")
def extract_ip(line):
    m = re.search(r"\d+\.\d+\.\d+\.\d+", line)
    return m.group() if m else None
 
# ✅ 내장 정규식 함수 (JVM 내 실행, 빠름)
df = df.withColumn("ip", F.regexp_extract("line", r"(\d+\.\d+\.\d+\.\d+)", 1))

# Apache 액세스 로그 파싱
pattern = r'(\S+) \S+ \S+ \[([^\]]+)\] "(\S+) (\S+) [^"]*" (\d{3}) (\d+)'
 
parsed = df.select(
    F.regexp_extract("line", pattern, 1).alias("ip"),
    F.regexp_extract("line", pattern, 2).alias("ts"),
    F.regexp_extract("line", pattern, 3).alias("method"),
    F.regexp_extract("line", pattern, 4).alias("path"),
    F.regexp_extract("line", pattern, 5).cast("int").alias("status"),
    F.regexp_extract("line", pattern, 6).cast("long").alias("bytes"))

위험 패턴: 중첩된 수량자 (a+)+,  (.*)*,  (\d+)+$
악성 입력: "aaaaaaaaaaaaaaaaaaaaaaaa!" 같은 거의-매칭
→ 백트래킹이 폭발 → 한 행 처리에 수초~수분 → 잡 멈춤

# ❌ 위험: 중첩 수량자
r"(\w+\s*)+"
 
# ✅ 안전: 구체적 문자클래스, 앵커
r"^\w+(\s\w+)*$"

from pyspark.ml.feature import RegexTokenizer, StopWordsRemover
 
# 정규화: 소문자 + 특수문자 제거
df = df.withColumn("clean",
    F.regexp_replace(F.lower("text"), r"[^\w\s가-힣]", " "))
 
# 토큰화 (정규식 기반)
tokenizer = RegexTokenizer(inputCol="clean", outputCol="tokens",
                           pattern=r"\s+", minTokenLength=2)
df = tokenizer.transform(df)
 
# 불용어 제거
remover = StopWordsRemover(inputCol="tokens", outputCol="filtered")
df = remover.transform(df)

# 제어문자·비인쇄 문자 제거
df = df.withColumn("clean",
    F.regexp_replace("text", r"[\x00-\x1F\x7F]", ""))
 
# 읽기 시 인코딩 지정 (깨짐 방지)
df = spark.read.option("encoding", "UTF-8").text("path")
 
# 깨진 행이 많으면 분리·격리 (별도 글 "중첩 반정형 데이터"의 quarantine 패턴)

# 관심 있는 로그만 먼저 필터 → 그 다음 비싼 파싱
errors = df.filter(F.col("line").rlike(r"\bERROR\b"))
parsed = errors.select(F.regexp_extract(...))