PySpark GraphFrames 로 엔티티 해소 — Connected Components 군집화

수학 시간에 배운 등가 관계를 떠올려 보세요. "A=B이고 B=C면 A=C" — 추이적 연결입니다. 퍼지 매칭으로 "A와 B가 같다", "B와 C가 같다"는 쌍을 잔뜩 찾았을 때, 최종적으로 필요한 건 쌍이 아니라 "A, B, C 는 한 사람" 이라는 그룹입니다. 이 추이적 연결을 푸는 것은 본질적으로 그래프의 연결 요소(Connected Components) 문제입니다.

수백 건이면 메모리에서 풀지만, 매칭 쌍이 수억 개라면 분산 그래프 처리가 필요합니다. 이 글은 PySpark + GraphFrames 로 대규모 엔티티 군집화를 구현하는 법과, 반복 알고리즘 특유의 함정을 정리합니다.

이 글에서 배우는 것

• 매칭 쌍(pair)에서 엔티티 그룹으로 가는 문제가 왜 그래프 Connected Components 인지
• GraphFrames 의 정점(vertices)·간선(edges) DataFrame 을 구성하는 방법
• 반복 알고리즘에서 체크포인트가 왜 필수인지 (lineage 폭발 방지)
• 괴물 군집(Monster Cluster) 사고의 원인과 탐지·방어 전략
• PageRank·Motif Finding 등 GraphFrames 의 다른 활용법

1. 왜 그래프 문제인가

발상의 전환이 핵심입니다. 매칭 쌍을 간선(edge) 으로, 레코드를 정점(vertex) 으로 바라보면, 같은 엔티티는 하나의 연결된 덩어리가 됩니다.

핵심: A-C 매칭을 직접 못 찾았어도, A-B-C 경로로 연결되면 같은 엔티티입니다. 이 추이적 묶음을 SQL 셀프조인 반복으로 풀면 비효율적이고 끝이 안 보입니다. Connected Components 알고리즘이 정확히 이걸 위한 도구입니다.

2. GraphFrames 준비

GraphFrames 는 Spark 에 기본 내장된 패키지가 아닙니다. 실행할 때 별도로 로드해야 합니다.

# spark-submit / pyspark 실행 시
# --packages graphframes:graphframes:0.8.x-spark3.5-s_2.12
 
from graphframes import GraphFrame
from pyspark.sql import functions as F

그래프는 정점 DataFrame(id 필수) 과 간선 DataFrame(src, dst 필수) 으로 만듭니다.

# 정점: 모든 레코드 id
vertices = records.select(F.col("id")).distinct()
 
# 간선: 퍼지 매칭으로 찾은 매칭 쌍 (양방향 불필요 — CC 는 무방향 취급)
edges = matches.select(
    F.col("id_a").alias("src"),
    F.col("id_b").alias("dst"))
 
g = GraphFrame(vertices, edges)

3. Connected Components 실행

준비가 됐다면 실행 자체는 간단합니다. 단, 반복 알고리즘이라 체크포인트를 반드시 먼저 설정해야 합니다.

# 반복 알고리즘이라 체크포인트 디렉터리가 필수
spark.sparkContext.setCheckpointDir("/tmp/graph-checkpoints")
 
cc = g.connectedComponents()
# 결과: id, component  (같은 component 값 = 같은 엔티티)
 
cc.show()
# +----+------------+
# | id | component  |
# +----+------------+
# | A  | 1           |
# | B  | 1           |
# | C  | 1           |
# | D  | 4           |
# | E  | 4           |

component 컬럼이 엔티티 ID 역할을 합니다. 같은 값이면 같은 엔티티입니다.

# 엔티티별로 묶어 통합 마스터 생성
entities = (cc
    .groupBy("component")
    .agg(F.collect_list("id").alias("member_ids"),
         F.count("*").alias("cluster_size")))

4. 체크포인트가 왜 필수인가

Connected Components 는 반복(iterative) 알고리즘입니다. Spark 는 기본적으로 모든 변환의 계보(lineage)를 추적하는데, 반복이 거듭될수록 이 계보가 끝없이 길어집니다. 그 결과 플래너가 느려지고 StackOverflow·OOM 이 납니다. 체크포인트는 중간 결과를 디스크에 잘라(truncate lineage) 이를 막습니다.

spark.sparkContext.setCheckpointDir("hdfs:///checkpoints/cc")  # 반드시 설정

체크포인트 디렉터리를 설정하지 않으면 connectedComponents() 가 경고/실패합니다. 반복 그래프 알고리즘에서 체크포인트는 선택이 아니라 필수입니다.

5. 가장 큰 함정 — 괴물 군집(Monster Cluster)

엔티티 해소에서 가장 무서운 사고가 거대 군집입니다. 다리 하나가 무너지면 두 섬이 연결되듯, 잘못된 매칭 하나가 서로 무관한 두 큰 덩어리를 연결하면 수백만 레코드가 하나의 엔티티로 합쳐집니다. 매칭 단계의 작은 오류가 군집 단계에서 재앙으로 증폭되는 것이죠.

이것은 그래프에서 한 간선이 다리(bridge) 역할을 할 때 생깁니다. 대응:

# 괴물 군집 탐지
suspects = entities.where("cluster_size > 1000").orderBy(F.desc("cluster_size"))
suspects.show()

실무 철칙: Connected Components 결과를 그대로 신뢰하지 말고, 군집 크기 분포를 반드시 점검하세요. 매칭 단계의 작은 오류가 군집 단계에서 재앙으로 증폭됩니다. precision 을 약간 희생해 recall 을 얻는 매칭은, 군집화에서는 위험합니다.

6. 성능 — 반복 알고리즘 스케일링

반복 알고리즘은 데이터가 클수록, 간선이 많을수록 오래 걸립니다. 미리 그래프를 희소하게 만드는 것이 속도와 품질 모두에 이롭습니다.

간선이 빽빽할수록 반복이 오래 걸립니다. 약한 매칭을 미리 쳐내 그래프를 희소하게 만드는 것이 속도와 품질 모두에 이롭습니다.

7. Connected Components 외 다른 그래프 알고리즘

GraphFrames 는 엔티티 해소만을 위한 도구가 아닙니다. 관계 데이터를 다루는 다양한 문제에 활용할 수 있습니다.

# Motif: A→B→C→A 순환 거래 탐지 (자금세탁 의심 패턴)
motifs = g.find("(a)-[]->(b); (b)-[]->(c); (c)-[]->(a)")

8. 전체 흐름 (퍼지 매칭과 연결)

9. 정리

엔티티 해소의 마지막 퍼즐은 "쌍을 군집으로 묶는" 그래프 단계입니다. GraphFrames 의 Connected Components 가 이를 분산으로 깔끔히 풀어주지만, 체크포인트를 잊지 말 것과 괴물 군집을 항상 경계할 것 두 가지가 성패를 가릅니다. 매칭과 군집은 한 몸이라, 매칭의 정밀도가 곧 군집의 안전성이라는 점을 기억하세요.

마치며 — 핵심 요약

• 매칭 쌍(A=B, B=C)을 엔티티 그룹(A=B=C)으로 바꾸는 것은 그래프 Connected Components 문제입니다.
• GraphFrames 는 정점 DataFrame(id)과 간선 DataFrame(src, dst)으로 그래프를 표현합니다.
• 반복 알고리즘에서 체크포인트 디렉터리 설정은 필수입니다. 없으면 lineage 폭발로 느려지거나 실패합니다.
• 괴물 군집은 오매칭 1건이 수백만 레코드를 한 엔티티로 합쳐버리는 재앙입니다. 결과를 그대로 신뢰하지 말고 cluster_size 분포를 반드시 점검하세요.
• 약한 매칭 간선을 미리 쳐내 그래프를 희소하게 만들면 속도와 품질이 모두 좋아집니다.
• 퍼지 매칭(이전 글)과 그래프 군집화(이 글)는 세트입니다. 둘을 함께 설계해야 전체 파이프라인이 안정적입니다.

이 글은 Spark 3.5 + GraphFrames 기준으로 작성되었습니다. 대규모 엔티티 해소·관계 분석·부정탐지 그래프 파이프라인이 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀