소개
소매업체에는 이미 올바른 자료형의 결제 완료 주문 결과가 있습니다. 이제 다른 애플리케이션이 나중에 이 결과를 읽고 날짜 하나를 선택하며 반복 내보내기로 주문이 조용히 중복되지 않았다고 믿을 수 있어야 합니다. 저장은 스크립트의 마지막 줄에 불과한 일이 아니라 처리 약속의 일부입니다.
order_date에 따른 **디렉터리 파티션(directory partitioning)**으로 Parquet를 씁니다. 데이터셋 루트에는 날짜 이름의 하위 디렉터리가 생기고 Spark는 루트를 읽을 때 날짜 열을 복원합니다. 이 저장 디렉터리는 작업 실행용 계산 파티션과 다릅니다. 다음 실습에서 실행을 따로 살펴봅니다.
이 실습은 십진수 금액이 있는 결제 완료 주문 네 건을 로컬 Parquet 입력으로 독립적으로 준비합니다. DataFrame 읽기, 필터와 최소 덮어쓰기 Parquet 작업 흐름을 알아야 합니다. Hive 카탈로그, HDFS, SQL 선수 지식, 설치나 이전 출력은 필요하지 않습니다. 터미널을 사용하세요.
날짜 파티션 데이터셋을 쓰고 다시 읽기
이 단계에서는 준비된 결과를 날짜 디렉터리로 정리합니다. data/prepared-orders 아래 입력에는 앞서 사용한 것과 같은 자료형의 수량, 십진수 가격, 날짜와 십진수 order_amount를 가진 주문 네 건이 있습니다. Parquet는 이진 형식이므로 cat으로 CSV 텍스트를 기대하지 말고 Spark로 확인하세요.
작업 공간에서 시작합니다.
cd /home/labex/project
partitionBy("order_date")는 쓰기 작업에 날짜 이름 디렉터리별로 행을 넣도록 지시합니다. mode("overwrite")는 지정 출력 경로의 기존 파생 데이터셋을 교체합니다. 출력 경로는 온전히 유지할 input_path와 분리하세요.
전체 프로그램을 저장하고 실행합니다.
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/partitioned-read-back.txt
cat /home/labex/project/output/partitioned-read-back.txt
복원된 네 기록이 예상됩니다. 10월 14일의 o1과 o2, 10월 15일의 o4와 o6이며 금액은 20.00, 25.00, 50.00, 30.00입니다. Main rows는 4여야 합니다. quantity가 integer, 가격과 금액이 decimal, order_date가 date인지 확인하세요.
Spark는 파티션 열을 각 파티션 파일의 일반 필드가 아닌 디렉터리 이름에서 복원하므로 읽은 뒤 열 위치가 달라질 수 있습니다. 열 이름과 값이 정체를 정합니다. 이후 소비자에게 특정 열 순서가 필요하면 명시적 select를 사용하세요. 표시 순서에도 orderBy가 필요합니다.
디렉터리를 확인하고 날짜 하나 선택하기
이 단계에서는 디렉터리 구조를 복원한 날짜 열과 연결합니다. 출력을 재귀적으로 나열합니다. -R은 날짜 하위 디렉터리와 그 파일을 포함합니다.
ls -R /home/labex/project/output/orders-by-date
order_date=2026-10-14와 order_date=2026-10-15 디렉터리가 예상됩니다. part 파일 이름, 개수와 숨겨진 메타데이터는 달라질 수 있습니다. 특정 part 파일 하나를 데이터셋으로 하드코딩하지 마세요. 의미 있는 루트는 orders-by-date입니다.
이 루트를 읽으면 Spark가 파티션 값을 발견합니다. 이 파일 구조 발견에는 Hive 파티션이나 테이블 등록이 필요하지 않습니다. 다음 독립 스크립트는 루트를 읽고 DateType 열을 유효한 ISO 날짜 문자열과 비교하며 선택한 날짜의 기록을 별도로 저장합니다.
cat > /home/labex/project/inspect_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("InspectDateDataset").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
root = "/home/labex/project/output/orders-by-date"
df = spark.read.parquet(root)
selected = df.filter(col("order_date") == "2026-10-14")
selected.orderBy("order_id").show(truncate=False)
selected.write.mode("overwrite").parquet("/home/labex/project/output/selected-day")
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_sales_data.py > /home/labex/project/output/selected-date.txt
cat /home/labex/project/output/selected-date.txt
10월 14일 날짜와 금액 20.00, 25.00을 가진 o1과 o2만 예상됩니다. 다른 날짜는 주 데이터셋에 여전히 있습니다. 필터링한 읽기는 반환 기록을 바꾸며 선택하지 않은 디렉터리를 삭제하지 않습니다.
저장 구조와 실제 쿼리 결과를 확인했습니다. 이 작은 예제의 소요 시간으로 어느 형식이나 구조가 더 빠르다고 판단할 수는 없습니다. 목적은 이후 리더가 쓸 수 있는 올바르고 발견 가능한 데이터셋입니다.
반복 덮어쓰기와 추가 쓰기를 안전하게 비교하기
이 단계에서는 올바른 주 데이터셋을 보존하고 별도 디렉터리에서 실험합니다. 일회용 개수를 예상하세요. 같은 네 행을 두 번 덮어쓰면 매번 네 행이고, 전체 사본을 추가하면 여덟 행이 됩니다. 추가 행은 새 주문이 아니라 중복 내보내기입니다.
전체 버전을 실행합니다. 주 경로는 덮어쓰기를 사용하고 의도적 추가 쓰기는 write-mode-demo에만 적용합니다.
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
practice_path = "/home/labex/project/output/write-mode-demo"
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("First overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("Second overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("append").partitionBy("order_date").parquet(practice_path)
print("After append rows:", spark.read.parquet(practice_path).count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
Main rows: 4, First overwrite rows: 4, Second overwrite rows: 4, After append rows: 8을 찾으세요. 일회용 데이터셋은 이제 의도적으로 주문마다 사본 두 개를 포함하여, 전체 스냅샷 추가가 주 내보내기에 적절하지 않은 이유를 보여줍니다.
저장한 프로그램을 다시 실행하여 전체 작업 흐름의 반복 가능성을 확인합니다.
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
같은 개수가 나타나야 합니다. 전체 실행마다 주 데이터셋을 다시 만들고 일회용 실험을 초기화한 뒤 의도적으로 추가하므로 열두 행이나 열여섯 행으로 늘지 않습니다. 추가 쓰기는 실제로 새롭고 겹치지 않는 데이터를 위한 별도 정책입니다. 이 완전한 파생 스냅샷에는 덮어쓰기를 선택했습니다.
원래 prepared-orders 데이터셋은 계속 사용할 수 있습니다. 날짜 디렉터리는 저장된 값을 설명하고 Spark 실행 시 계산 파티션은 작업이 나뉘는 방식을 설명합니다. 다음에는 파티션의 두 번째 의미를 계획과 실제 실행에 연결합니다.
요약
날짜 파티션 Parquet 데이터셋을 쓰고 루트에서 자료형이 있는 기록을 복원했으며 다른 날짜를 제거하지 않고 날짜 하나를 선택했습니다. 반복 덮어쓰기는 주 스냅샷을 보존했고 일회용 추가 쓰기는 중복 사본을 보여주었습니다. 다음에는 실행 계획과 실제 이벤트로 계산 파티션, 작업, 스테이지와 태스크를 설명합니다.
