JSON 이벤트를 Spark DataFrame으로 읽기

소개

어떤 시스템은 기록을 CSV 행으로 보내고 다른 시스템은 이름 있는 필드의 JSON 객체로 보냅니다. 이제 소매업체는 유형 표시와 주문 관련 값이 있는 이벤트 파일을 제공합니다. 처리할 질문은 익숙합니다. 필드가 무엇을 뜻하는지, 어떤 기록이 주문인지, 결과에서 어떤 열을 보여줄지입니다.

JSON Lines는 한 줄에 완전한 JSON 객체 하나를 저장합니다. CSV와 달리 각 객체에 필드 이름이 있고 객체 안에 다른 객체를 넣을 수 있습니다. 먼저 명시적 자료형으로 평평한 기록을 읽고, 간단한 중첩 고객 객체 하나의 필드를 선택한 뒤 그 스키마를 두 번째 데이터에 적용합니다.

spark.read로 읽는 유한한 파일이므로 여전히 **배치 처리(batch processing)**입니다. 파일 이름에 events가 있다고 애플리케이션이 스트리밍이 되지는 않습니다. 환경과 데이터는 독립적입니다. 앞서 배운 명시적 스키마, select, filter, alias와 기본 Python이 필요합니다. Kafka, Hadoop, Hive, 스트리밍 설정, 배열이나 복잡한 재귀 구조는 필요하지 않습니다. 터미널을 사용하세요.

평평한 JSON Lines를 읽고 주문 이벤트 선택하기

이 단계에서는 명시적 스키마 읽기를 CSV에서 JSON으로 옮깁니다. 제공된 파일을 살펴봅니다.

cd /home/labex/project

cat /home/labex/project/data/flat-events.jsonl

각 줄은 중괄호로 둘러싼 객체입니다. 큰따옴표로 묶은 이름은 필드를 식별하고 콜론은 이름과 값을 나눕니다. quantity 같은 숫자는 JSON 숫자 값이고 order_date는 선언한 DateType이 해석할 따옴표로 묶인 날짜 문자열입니다. CSV 헤더 줄이나 바깥 배열은 없습니다.

첫 객체의 order_id, 수량, 가격과 날짜는 CSV 주문과 같은 종류의 값을 설명합니다. 파일에는 재고 이벤트도 있습니다. 유효한 기록을 읽었다고 해서 주문 결과에 속하는 것은 아닙니다. 아래 명시적 스키마는 평평한 필드를 선언합니다. json(path)가 형식을 읽고 익숙한 filter가 event_type == order를 선택합니다.

따옴표로 묶은 히어 도큐먼트로 전체 스크립트를 저장하고 준비된 PySpark 인터프리터로 실행합니다.

cat > /home/labex/project/read_json_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
    StructField("event_type", StringType(), True),
    StructField("order_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_date", DateType(), True),
])
events = (
    spark.read.schema(event_schema)
    .option("dateFormat", "yyyy-MM-dd")
    .json("/home/labex/project/data/flat-events.jsonl")
)
events.printSchema()
events.orderBy("order_id").show(truncate=False)
orders = events.filter(col("event_type") == "order")
print("Order events:")
orders.select("order_id", "quantity", "unit_price", "order_date").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON

bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/flat-json-orders.txt

cat /home/labex/project/output/flat-json-orders.txt

처음 표시에는 이벤트 기록 세 개 모두가 있어야 합니다. Order events 아래에는 j1/2/10.00/2026-10-14와 j2/1/25.00/2026-10-14만 예상됩니다. 재고 기록은 성공적으로 읽었지만 유형 조건으로 제외되었습니다. 스키마는 수량 integer, 가격 decimal, order_date date를 보여야 합니다.

리더는 이름 있는 JSON 필드에 선언한 자료형을 적용합니다. 일반 CSV 기록과 달리 각 JSON 객체 안의 필드 순서가 열의 정체를 정하지는 않습니다. 스키마는 여전히 의도한 약속을 정의합니다.

중첩 고객 객체의 필드 선택하기

이 단계에서는 구조 수준을 하나 더 읽습니다. 고객 이벤트 파일을 확인합니다.

cat /home/labex/project/data/customer-events.jsonl

customer 값은 customer_id와 region을 포함하는 또 다른 객체입니다. Spark 스키마에서는 이름 있는 하위 필드를 담는 struct가 됩니다. 배열이나 조인할 두 번째 테이블이 아닙니다.

안쪽 StructType으로 struct를 선언합니다. col("customer.region")은 점으로 구분한 경로로 하위 필드를 선택하고 alias("region")은 선택 결과에 간단한 열 이름을 줍니다. 고객 식별자에도 같은 개념을 적용합니다.

cat > /home/labex/project/read_json_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
    StructField("event_type", StringType(), True),
    StructField("order_id", StringType(), True),
    StructField("customer", StructType([
        StructField("customer_id", StringType(), True),
        StructField("region", StringType(), True),
    ]), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_date", DateType(), True),
])
events = (
    spark.read.schema(event_schema)
    .option("dateFormat", "yyyy-MM-dd")
    .json("/home/labex/project/data/customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
    "order_id",
    col("customer.customer_id").alias("customer_id"),
    col("customer.region").alias("region"),
    "quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON

bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/nested-json-orders.txt

cat /home/labex/project/output/nested-json-orders.txt

전체 스키마는 customer를 문자열 하위 필드 두 개가 있는 struct로 보여야 합니다. 선택한 테이블은 j3/c1/East/2/10.00/2026-10-15와 j4/c2/West/1/25.00/2026-10-15여야 합니다. 최종 스키마는 수량, 가격, 날짜와 함께 customer_id와 region을 일반 문자열 열로 제공합니다.

각 주문 이벤트 안에 이미 있던 값을 선택했습니다. 조회 테이블이나 조인은 필요하지 않았습니다. events에는 원래 중첩 표현이 남고 selected는 이후 연산에 더 단순한 평평한 보기를 제공합니다.

다른 파일에 중첩 스키마 재사용하기

이 단계에서는 구조를 유지하면서 데이터를 바꿉니다. 먼저 새 파일을 읽습니다.

cat /home/labex/project/data/new-customer-events.jsonl

출력을 예상해 보세요. 10월 16일 West의 c2 고객 주문 j5 한 건이며 수량 3, 가격 12.50입니다. 리더와 중첩 필드 선택은 같고 입력 경로만 바뀝니다. 전체 버전을 실행합니다.

cat > /home/labex/project/read_json_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
    StructField("event_type", StringType(), True),
    StructField("order_id", StringType(), True),
    StructField("customer", StructType([
        StructField("customer_id", StringType(), True),
        StructField("region", StringType(), True),
    ]), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_date", DateType(), True),
])
events = (
    spark.read.schema(event_schema)
    .option("dateFormat", "yyyy-MM-dd")
    .json("/home/labex/project/data/new-customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
    "order_id",
    col("customer.customer_id").alias("customer_id"),
    col("customer.region").alias("region"),
    "quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON

bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/new-json-orders.txt

cat /home/labex/project/output/new-json-orders.txt

예상한 한 행 결과와 같은 선택 자료형이 있어야 합니다. 합의한 구조가 같다면 새 파일이라고 새 스키마를 만들어낼 필요는 없습니다.

세 읽기 모두 기존 파일에 spark.read를 사용했으며 스크립트가 완료되면 끝났습니다. 앞으로 올 메시지를 구독하지 않았습니다. 실행 모델까지 동시에 바꾸지 않고 입력 형식을 바꾸는 방법을 배웠습니다. 다음에는 의도적으로 불완전한 주문 데이터로 돌아가 명시적인 정제 규칙을 정의합니다.

요약

명시적 스칼라 자료형으로 정적 JSON Lines를 읽고 주문 이벤트를 선택했으며 이름 있는 열 경로로 간단한 고객 struct를 평탄화했습니다. 두 번째 데이터에도 같은 스키마를 올바르게 사용했습니다. 다음에는 명확한 정제 규칙에 따라 누락, 중복 및 잘못된 주문 기록을 처리합니다.