누락, 중복 및 잘못된 PySpark 주문 정제하기

소개

스크립트가 성공적으로 실행되어도 잘못된 기록을 보고서에 전달할 수 있습니다. 오늘 받은 상품 주문에는 반복된 주문 행, 누락된 고객, 잘못된 숫자 텍스트, 양수가 아닌 수량, 음수 가격과 잘못된 날짜가 있습니다. 이런 문제를 자동으로 0으로 채우면 업무 정보를 지어내게 됩니다.

원래 증거를 잃지 않고 텍스트 읽기, 의도한 숫자 및 날짜 필드의 안전한 해석, 명시된 규칙에 맞는 행만 유지하기라는 세 결정을 구분합니다. 그다음 동일한 유효 행의 중복을 제거합니다. 마지막으로 정제된 Parquet 데이터셋을 저장하고 다시 읽습니다. Parquet는 자료형이 있는 이진 열 기반 파일 형식입니다. 이번에는 최소 쓰기 및 읽기 흐름만 배우고 파티션 저장은 이후에 다룹니다.

준비된 환경은 독립적인 데이터 두 벌을 제공합니다. CSV 읽기, Spark 열 표현식, 필터링과 십진수 가격을 알아야 합니다. 여기의 안전한 변환 표현식은 준비된 Spark 버전에서 지원되며 오류 처리 구성을 바꿀 필요가 없습니다. 터미널을 사용합니다. 이전 출력 파일, Hadoop, Hive나 서비스 설치는 필요하지 않습니다.

문자열 스테이징 데이터로 원본 보존하기

이 단계에서는 값을 변환하기 전에 결함을 관찰합니다. 로컬 원본을 살펴봅니다.

cd /home/labex/project

cat /home/labex/project/data/dirty-orders.csv

업무 행은 열네 개입니다. 스테이징(staging) DataFrame은 입력 필드를 문자열로 유지하므로 two, unknown, not-a-date 같은 값을 유효한 숫자나 날짜인 것처럼 다루지 않고 볼 수 있습니다. CSV 리더의 기본 처리는 빈 고객 필드를 NULL로 만듭니다. NULL은 누락값이며 텍스트 "NULL"이 아닙니다.

전체 스테이징 프로그램을 저장하고 실행합니다. 앞에서처럼 따옴표로 묶은 PYTHON 히어 도큐먼트가 소스를 그대로 씁니다. show(30, truncate=False)는 최대 서른 기록을 표시하고 필드 텍스트를 읽기 쉽게 유지합니다.

cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr

spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
staging.printSchema()
staging.orderBy("order_id").show(30, truncate=False)
print("Staging rows:", staging.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/staging-orders.txt

cat /home/labex/project/output/staging-orders.txt

표시 기록 열네 개와 Staging rows: 14가 예상됩니다. 모든 열이 문자열이어야 합니다. b2의 수량 two와 b4의 가격 unknown을 찾으세요. 추측한 기본값으로 숨길 오류가 아니라 유용한 증거입니다. o2는 두 번 나오고 o1의 상태에는 주변 공백이 있습니다.

이번 연습의 원본 파일은 읽기 전용입니다. 원본을 검토할 수 있게 두면서 파생 DataFrame과 출력 디렉터리를 만듭니다.

기본값을 지어내지 않고 안전하게 해석하기

이 단계에서는 주변 공백을 제거해야 하는 텍스트 필드 두 개를 정규화하고 숫자 및 날짜 변환을 시도합니다. trim은 앞뒤 공백을 제거합니다. 누락된 고객을 추측하거나 주문 결제 상태를 바꾸지는 않습니다.

expr는 텍스트로 Spark 내장 표현식을 제공합니다. try_cast(quantity AS INT)는 정수 변환을 시도하고 변환할 수 없으면 NULL을 만듭니다. 같은 방식으로 가격에는 DECIMAL(10,2), 날짜에는 DATE를 요청합니다. 일반 변환 실패는 Spark의 ANSI 동작에서 오류를 낼 수 있습니다. 이 명시적인 안전한 변환은 전역 설정에 의존하지 않습니다.

해석 문제가 보이도록 스키마가 바뀌어도 행은 유지합니다. 이 버전을 실행합니다.

cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr

spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
    staging
    .withColumn("customer_id", trim(col("customer_id")))
    .withColumn("order_status", trim(col("order_status")))
    .withColumn("quantity", expr("try_cast(quantity AS INT)"))
    .withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
    .withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
parsed.printSchema()
parsed.orderBy("order_id").show(30, truncate=False)
print("Parsed rows:", parsed.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/parsed-orders.txt

cat /home/labex/project/output/parsed-orders.txt

Parsed rows: 14가 예상됩니다. b2의 수량, b4의 가격, b6의 날짜는 이제 NULL이어야 합니다. b3의 수량 0과 b5의 음수 가격은 문법상 유효한 숫자이지만 다음에 적용할 업무 규칙을 충족하지 않습니다. o1의 상태는 주변 공백 없는 paid여야 합니다.

변환 성공과 유효한 주문은 다릅니다. 단계를 구분하면 행이 해석에 실패했는지 업무 제약을 위반했는지 설명할 수 있습니다. 누락 가격을 0으로 바꾸지 않았으므로 실제 가격이 0인 o7과 알 수 없는 가격을 구별할 수 있습니다.

명시적인 유효성 및 중복 규칙 적용하기

이 단계에서는 유효한 업무 기록을 유지하고 완전히 같은 중복만 제거합니다. 다음 규칙을 사용합니다.

customer_id가 존재하고 공백 제거 후 비어 있지 않습니다.
quantity가 성공적으로 해석되었고 0보다 큽니다.
unit_price가 성공적으로 해석되었고 0 이상입니다.
order_date가 성공적으로 해석되었습니다.
정규화한 일곱 필드 전체가 같은 행은 하나만 유지합니다.

예제의 나머지 식별자 필드는 이미 올바릅니다. 같은 주문 ID에 충돌하는 버전이 없으므로 바뀌는 상태 중 어느 것을 선택할지는 다루지 않습니다. 다른 조건이 유효하면 결제 완료, 대기, 환불 기록을 모두 유지합니다. 데이터 정제와 보고서용 결제 완료 매출 선택은 다른 일입니다.

isNotNull은 존재를 명시적으로 검사합니다. NULL이 포함된 비교는 참이 되지 않으며 filter는 참인 조건만 유지합니다. &로 조건을 결합하면 정책이 드러납니다. 부분집합 없이 dropDuplicates()를 사용하면 완전히 같은 행을 제거합니다. order_id만 사용하면 충돌 정보를 버릴 수 있는 다른 정책이 됩니다.

이 버전을 실행합니다.

cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr

spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
    staging
    .withColumn("customer_id", trim(col("customer_id")))
    .withColumn("order_status", trim(col("order_status")))
    .withColumn("quantity", expr("try_cast(quantity AS INT)"))
    .withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
    .withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
    col("customer_id").isNotNull()
    & (col("customer_id") != "")
    & col("quantity").isNotNull() & (col("quantity") > 0)
    & col("unit_price").isNotNull() & (col("unit_price") >= 0)
    & col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
print("Clean rows:")
clean.orderBy("order_id").show(30, truncate=False)
print("Staging count:", staging.count())
print("Valid count before deduplication:", valid.count())
print("Clean count:", clean.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/cleaning-result.txt

cat /home/labex/project/output/cleaning-result.txt

Staging count: 14, Valid count before deduplication: 8, Clean count: 7이 예상됩니다. b1–b6은 명시한 규칙으로 제외되고 반복된 o2 하나가 제거됩니다. 유지된 일곱 ID는 o1부터 o7입니다.

o7의 실제 가격 0은 음수가 아닌 가격 규칙에 유효합니다. o3은 여전히 대기이고 o5는 환불입니다. 어느 상태도 원본 기록을 잘못된 형식으로 만들지는 않습니다. 결과 데이터셋은 명시한 정책에 따라 정제되었으며 모든 현실의 이상 상황을 해결했다는 주장은 아닙니다.

저장하고 다시 읽으며 추가 결함 테스트하기

이 단계에서는 정제 데이터셋을 재사용 가능하게 만들고 다른 데이터에 같은 정책을 테스트합니다. write.mode("overwrite").parquet(path)는 디렉터리에 Parquet 데이터셋을 쓰고 이전 파생 내용을 교체합니다. 그 디렉터리는 원본 입력과 분리하세요. Spark는 여러 part 파일과 완료 표시를 만들 수 있으므로 특정 파일 이름이나 개수에 의존하지 마세요.

spark.read.parquet(path)는 저장된 데이터셋과 자료형을 읽습니다. 쓰기 전 계획만 표시하는 대신 다시 읽은 DataFrame을 조회하면 실제 결과물을 검사합니다. 아래 전체 프로그램은 이미 배운 로직을 Python 함수에 두어 두 데이터에 같은 정제 규칙을 적용합니다.

def clean_orders(input_path):는 함수를 정의하고 입력 매개변수에 이름을 줍니다. 들여쓴 줄은 함수 본문이며 호출될 때 실행됩니다. 첫 호출은 원래 파일 이름을, 두 번째 호출은 추가 데이터 파일 이름을 제공합니다. 각 호출 안의 input_path는 제공한 값을 가리킵니다. return clean은 결과 DataFrame을 호출자에 반환하며 파일을 저장하지 않습니다. 들여쓰지 않은 쓰기 줄이 별도의 저장 액션을 수행합니다. 전체 프로그램을 복사할 때 들여쓰기를 유지하세요.

추가 데이터에는 원래 열네 행에 음수 수량 행과 유효한 가격 0 주문의 정확한 사본이 하나씩 더 있습니다. 정제 결과를 예상해 보세요. 이 추가로 유지되는 일곱 기록은 바뀌지 않아야 합니다. 추가 항목을 확인합니다.

tail -n 2 /home/labex/project/data/dirty-orders-extra.csv

전체 저장 및 다시 읽기 프로그램을 실행합니다.

cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr

spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

def clean_orders(input_path):
    staging = spark.read.option("header", True).csv(input_path)
    parsed = (
        staging
        .withColumn("customer_id", trim(col("customer_id")))
        .withColumn("order_status", trim(col("order_status")))
        .withColumn("quantity", expr("try_cast(quantity AS INT)"))
        .withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
        .withColumn("order_date", expr("try_cast(order_date AS DATE)"))
    )
    valid = parsed.filter(
        col("customer_id").isNotNull()
        & (col("customer_id") != "")
        & col("quantity").isNotNull() & (col("quantity") > 0)
        & col("unit_price").isNotNull() & (col("unit_price") >= 0)
        & col("order_date").isNotNull()
    )
    clean = valid.dropDuplicates()
    return clean

clean = clean_orders("/home/labex/project/data/dirty-orders.csv")
clean.write.mode("overwrite").parquet("/home/labex/project/output/clean-orders")
reloaded = spark.read.parquet("/home/labex/project/output/clean-orders")
print("Read-back clean rows:")
reloaded.orderBy("order_id").show(30, truncate=False)
reloaded.printSchema()
print("Read-back count:", reloaded.count())

practice = clean_orders("/home/labex/project/data/dirty-orders-extra.csv")
practice.write.mode("overwrite").parquet("/home/labex/project/output/practice-clean-orders")
practice_reloaded = spark.read.parquet("/home/labex/project/output/practice-clean-orders")
print("Extra-delivery clean rows:")
practice_reloaded.orderBy("order_id").show(30, truncate=False)
print("Extra-delivery count:", practice_reloaded.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/clean-parquet-check.txt

cat /home/labex/project/output/clean-parquet-check.txt

다시 읽은 두 테이블 모두 o1부터 o7이 한 번씩 있고 개수는 모두 7이어야 합니다. 주 출력 디렉터리는 output/clean-orders이며 practice-clean-orders는 별도의 파생 실험입니다. 다시 읽은 스키마는 수량 integer, 가격 decimal과 order_date date를 유지해야 합니다.

이것은 최소 Parquet 저장 및 읽기이며 속도 비교나 저장소 튜닝 연습이 아닙니다. 이제 다른 Spark 프로세스가 읽을 수 있는 작은 정제 데이터셋과, 새 잘못된 행 및 중복이 결과를 조용히 바꾸지 않는다는 증거가 있습니다.

요약

원본 텍스트를 보존하고 안전한 변환으로 잘못된 값을 드러냈으며 명시적인 누락값 및 범위 정책을 적용하고 완전히 같은 중복 행을 제거했습니다. 유효한 가격 0과 미결제 기록을 유지한 뒤 실제 자료형이 있는 Parquet 출력과 두 번째 데이터를 검증했습니다. 다음 도전 과제에서는 이 정제 및 금액 계산 기술을 독립적으로 적용합니다.

누락, 중복 및 잘못된 Spark 주문 기록 정제하기

소개

문자열 스테이징 데이터로 원본 보존하기

기본값을 지어내지 않고 안전하게 해석하기

명시적인 유효성 및 중복 규칙 적용하기

저장하고 다시 읽으며 추가 결함 테스트하기

요약