PySpark 깊게 중첩된 반정형 데이터 다루기 — 5단계 JSON과 스키마 진화

택배 상자를 열었더니 또 다른 상자가, 그 안을 열었더니 또 포장된 물건들이 가득한 상황을 상상해 보세요. API 응답과 이벤트 로그가 딱 그렇습니다. 5단계로 중첩된 JSON, 배열 안에 또 객체가 든 구조, 게다가 소스 버전마다 필드가 들쭉날쭉합니다. 이런 반정형(semi-structured) 데이터를 분석 가능한 형태로 길들이는 것은 데이터 엔지니어링의 일상적이지만 까다로운 과제입니다.

이 글은 PySpark 로 깊게 중첩된 데이터를 다루는 핵심 도구 — struct 접근, explode, from_json, 평탄화, 스키마 진화 — 를 실전 패턴으로 정리합니다.

이 글에서 배우는 것

• Spark 의 복합 타입(StructType, ArrayType, MapType)을 어떻게 구분하고 다루는지
• 점 표기와 재귀 평탄화로 중첩 struct 를 펼치는 방법
• explode / explode_outer / posexplode 차이와 빈 배열 함정
• Kafka·로그처럼 JSON 문자열로 들어오는 데이터를 from_json 으로 파싱하는 법
• 깨진 레코드를 조용히 버리지 않고 격리(quarantine)하는 패턴과 스키마 진화 대응

1. 중첩 데이터의 구조 이해

도구를 꺼내기 전에 지도부터 펴야 합니다. Spark 의 복합 타입은 세 가지뿐이고, 이것만 구분하면 어떤 중첩 구조든 분해할 수 있습니다.

df.printSchema()
# root
#  |-- user: struct
#  |    |-- id: long
#  |    |-- profile: struct
#  |    |    |-- name: string
#  |    |    |-- tags: array<string>
#  |-- events: array<struct<type:string, ts:long>>

2. Struct 접근 — 점 표기와 평탄화

printSchema() 로 구조를 확인했다면, 이제 원하는 필드를 꺼낼 차례입니다. 파이썬 딕셔너리 키 접근과 비슷하게, 중첩 struct 필드는 점(.)으로 접근하면 됩니다.

from pyspark.sql import functions as F
 
df.select(
    F.col("user.id").alias("user_id"),
    F.col("user.profile.name").alias("name"),
)

struct 전체 평탄화(flatten)

필드를 하나씩 고르는 대신, 깊은 struct 를 통째로 펼치고 싶을 때가 있습니다 — BI 도구에 넘기거나 탐색적 분석을 할 때가 대표적입니다. 재귀적으로 필드를 순회하면 됩니다.

def flatten(df):
    """모든 struct 필드를 최상위 컬럼으로 평탄화"""
    flat_cols = []
    nested = []
    for field in df.schema.fields:
        if isinstance(field.dataType, T.StructType):
            nested.append(field.name)
        else:
            flat_cols.append(F.col(field.name))
    for n in nested:
        for child in df.schema[n].dataType.fields:
            flat_cols.append(F.col(f"{n}.{child.name}").alias(f"{n}_{child.name}"))
    flat = df.select(flat_cols)
    # 아직 struct 가 남아있으면 재귀
    if any(isinstance(f.dataType, T.StructType) for f in flat.schema.fields):
        return flatten(flat)
    return flat

평탄화는 분석/BI 친화적이지만, 깊은 구조를 다 펼치면 컬럼이 수백 개가 될 수 있습니다. 필요한 필드만 select 하는 것이 보통 더 낫습니다 — 전체 평탄화는 탐색 단계에만.

3. Explode — 배열을 행으로

struct 는 점으로 파고들었는데, 배열은 다릅니다. 배열 하나를 여러 행으로 "폭발(explode)" 시켜야 합니다. 한 행이 가진 N개 원소가 N개의 독립 행이 되는 것이죠.

한 문장으로: explode 는 가로로 긴 배열을 세로로 늘어선 행들로 바꿉니다.

# events 배열을 행으로
exploded = df.select(
    "user.id",
    F.explode("events").alias("event"))  # 배열 → 행
 
# 펼친 뒤 struct 필드 접근
result = exploded.select(
    "id",
    F.col("event.type").alias("event_type"),
    F.col("event.ts").alias("ts"))

# 빈 배열도 보존해야 하면 (left join 같은 효과)
F.explode_outer("events")
 
# 순번이 필요하면
df.select("id", F.posexplode("events").alias("pos", "event"))

흔한 버그: explode 를 쓰면 빈 배열·NULL 행이 조용히 사라집니다. 원본 행 수를 보존해야 한다면 explode_outer 를 쓰세요.

배열 속의 배열 (중첩 explode)

# orders[].items[] — 두 번 explode
df.select("order_id", F.explode("orders").alias("order")) \
  .select("order_id", F.explode("order.items").alias("item")) \
  .select("order_id", "item.sku", "item.qty")

4. JSON 문자열 파싱 — from_json

Kafka 메시지나 로그 컬럼처럼, 데이터가 Spark 에 이미 struct 가 아닌 JSON 문자열 그대로 들어오는 경우가 많습니다. 이런 컬럼은 겉으로는 string 타입이라 점 표기도, explode 도 바로 쓸 수 없습니다. from_json 으로 먼저 struct 로 변환해야 합니다.

from pyspark.sql import types as T
 
# 스키마 명시 (권장 — 안정적, 빠름)
schema = T.StructType([
    T.StructField("id", T.LongType()),
    T.StructField("tags", T.ArrayType(T.StringType())),
    T.StructField("meta", T.StructType([
        T.StructField("lang", T.StringType()),
    ])),
])
 
parsed = df.withColumn("data", F.from_json("json_str", schema))
result = parsed.select("data.id", "data.tags", "data.meta.lang")

스키마를 명시하지 않고 추론(schema_of_json)에 의존하면 느리고 불안정합니다. 가능하면 스키마를 명시하세요. 미지의 필드가 많은 탐색 단계에서만 추론을 쓰고, 프로덕션은 고정 스키마로.

5. 깨진 레코드 처리 — 안전하게

완벽한 JSON 만 들어오면 좋겠지만, 현실은 그렇지 않습니다. 닫히지 않은 중괄호, 예상치 못한 타입, 잘린 메시지 — 실전 JSON 에는 깨진 레코드가 반드시 섞입니다. 무방비로 읽으면 잡 전체가 실패하거나, 한 행 때문에 파티션 전체가 NULL 로 오염됩니다.

⚠️ 깨진 레코드를 조용히 버리면 데이터 유실이 생겨도 눈치채지 못합니다. 격리가 정답입니다.

# 읽기 모드: PERMISSIVE(기본), DROPMALFORMED, FAILFAST
df = (spark.read
    .option("mode", "PERMISSIVE")                       # 깨진 건 _corrupt_record 로
    .option("columnNameOfCorruptRecord", "_corrupt")
    .schema(schema)
    .json("path"))
 
# 깨진 레코드 분리 (버리지 말고 격리 → 나중에 분석)
good = df.where("_corrupt IS NULL")
bad  = df.where("_corrupt IS NOT NULL")
bad.write.mode("append").parquet("quarantine/")

권장: 깨진 레코드를 버리지 말고 격리(quarantine) 하세요. 조용히 버리면 데이터 유실을 눈치채지 못합니다. 격리 테이블에 쌓아두면 소스 문제를 추적할 수 있습니다.

6. 스키마 진화(Schema Evolution)

서비스가 자라면 API 응답에 새 필드가 생기고, 오래된 필드가 사라집니다. 어제 쌓인 Parquet 과 오늘 쌓인 Parquet 의 스키마가 다른 상황은 피할 수 없죠. PySpark 와 Lakehouse 포맷은 이런 스키마 진화를 자연스럽게 흡수할 수 있습니다. 소스 버전마다 필드가 추가/삭제될 때 쓸 수 있는 방법은 다음과 같습니다.

# 읽기 시 스키마 병합 (Parquet)
df = spark.read.option("mergeSchema", "true").parquet("path")
 
# 여러 DataFrame union 시 누락 컬럼 허용
combined = df_v1.unionByName(df_v2, allowMissingColumns=True)

Lakehouse 포맷은 스키마 진화를 테이블 차원에서 지원합니다.

# Iceberg/Delta: 쓰기 시 새 컬럼 자동 병합
(df.writeTo("analytics.events")
   .option("mergeSchema", "true")
   .append())

• 새 컬럼 추가는 안전합니다(기존 데이터는 NULL).
• 타입 변경·컬럼 삭제는 위험하니 정책을 정하세요. (Iceberg 스키마 진화 규칙은 별도 글 "Apache Iceberg 사양 버전 비교"에서 다룹니다.)

7. 성능 주의

중첩 데이터를 다루다 보면 행 수가 갑자기 수십 배 불어나거나, 전체 파일 스캔이 발생해 쿼리가 느려지는 상황을 만납니다. 미리 알아두면 피할 수 있는 함정들입니다.

explode 는 행을 늘리므로, 필터를 explode 전에 적용해 펼칠 데이터를 먼저 줄이는 것이 좋습니다.

8. 정리

복잡한 중첩 데이터의 핵심은 "구조를 먼저 이해하고(printSchema), 필요한 만큼만 펼치는 것"입니다. struct 는 점으로, 배열은 explode 로, JSON 문자열은 명시 스키마 from_json 으로 다루되, 빈 배열 유실(explode vs explode_outer) 과 깨진 레코드 격리라는 두 함정만 조심하면 됩니다. 무작정 전체 평탄화하기보다, 분석에 필요한 경로만 골라 펼치는 절제가 성능과 가독성 모두를 지킵니다.

마치며 — 핵심 요약

• 타입부터 확인: printSchema() 로 StructType / ArrayType / MapType 을 구분하면 어떤 도구를 쓸지 바로 보입니다.
• struct 는 점으로: col("user.profile.name") 처럼 경로를 이어 쓰고, 전체 평탄화는 탐색 단계에서만 — 프로덕션은 필요한 필드만 선택하세요.
• explode 의 함정: 빈 배열·NULL 이 있으면 해당 행이 조용히 사라집니다. 행 수를 보존해야 하면 반드시 explode_outer 를 쓰세요.
• JSON 문자열엔 명시 스키마: from_json 에 스키마를 직접 지정해야 빠르고 안정적입니다. schema_of_json 추론은 탐색용으로만.
• 깨진 레코드는 격리: PERMISSIVE 모드 + quarantine 테이블로 분리하세요. 조용히 버리면 데이터 유실을 뒤늦게야 발견합니다.
• 스키마 진화는 추가만 안전: 새 컬럼 추가는 mergeSchema 로 흡수할 수 있지만, 타입 변경·삭제는 별도 정책이 필요합니다.

중첩 데이터는 처음엔 복잡해 보여도, 타입 세 가지와 도구 세 가지(점 표기, explode, from_json)를 익히면 어떤 구조든 차분히 분해할 수 있습니다. 여러분의 다음 JSON 파이프라인에 한 번 적용해 보세요!

이 글은 Spark 3.5 기준으로 작성되었습니다. 반정형 데이터 수집·정규화 파이프라인이나 스키마 진화 전략 설계가 필요하시면 언제든 문의해 주세요.

— Data Dynamics 엔지니어링 팀