데이터 정제 및 품질 검사

Databricks에서 Spark SQL로 데이터 변환하기

Disha Mukherjee

Lead Data Engineer

실제 데이터의 세 가지 문제

recraft: half: 오류와 누락이 있는 문서와 파일 더미, 투명 배경

주요 컬럼의 결측값
반복 수집으로 인한 중복 행
논리적으로 유효하지 않은 레코드

스키마를 명시적으로 정의하는 이유

recraft: half: 스프레드시트 행 위에 물음표가 떠 있는 혼란스러운 과학자, 투명 배경

Spark가 컬럼 타입을 잘못 읽을 수 있습니다
숫자 컬럼이 문자열로 추론될 수 있습니다
StructType을 사용하면 매 실행 시 타입을 명시적으로 정의합니다

임포트

from pyspark.sql.types import (
    StructType, StructField,
    StringType, DoubleType, IntegerType, TimestampType
)


from pyspark.sql import functions as F

스키마 정의

schema = StructType([
    StructField("ID",                 IntegerType(),   True),
    StructField("Date",               TimestampType(), True),
    StructField("Transaction_Amount", DoubleType(),    True),
    StructField("Transaction_Status", StringType(),    True),
    # ... 6 more fields
])


df = (spark.read.format("csv")
      .option("header", "true")
      .schema(schema)
      .load("/Volumes/.../transactions.csv"))


df.printSchema()

 |-- ID: integer (nullable = true)
 |-- Date: timestamp (nullable = true)
 |-- Transaction_Amount: double (nullable = true)
 |-- Transaction_Status: string (nullable = true)
 ...

결측값 찾기

null_counts = df.select([
    F.sum(F.col(c).isNull().cast("int")).alias(c)
    for c in df.columns
])


null_counts.show()

+-----------+------------------+--------+--------+
|Customer_ID|Transaction_Amount|Category|Location|
+-----------+------------------+--------+--------+
|30         |50                |40      |30      |
+-----------+------------------+--------+--------+

결측값 제거 및 채우기

df_no_nulls = df.na.drop(subset=["Customer_ID"])


df_filled = df_no_nulls.na.fill({
    "Transaction_Amount": 0.0,
    "Category": "Unknown",
    "Location": "Unknown"
})


print(f"Before: {df.count()}")
print(f"After:  {df_filled.count()}")

Before: 100,150
After:  100,120

중복 처리

total = df_filled.count()
distinct = df_filled.distinct().count()

print(f"Duplicates: {total - distinct}")

Duplicates: 149

중복 처리

df_deduped = df_filled.dropDuplicates(
    subset=["Customer_ID", "Date", "Transaction_Amount"]
)


print(f"Rows after dedup: {df_deduped.count()}")

Rows after dedup: 99,971

유효하지 않은 레코드 필터링

df_valid = df_deduped.filter(F.col("Transaction_Amount") > 0)


df_valid = df_valid.filter(
    F.col("Transaction_Status") == "Completed"
)


print(f"Valid transactions: {df_valid.count()}")

Valid transactions: 33,223

파생 컬럼 생성

df_enriched = df_valid.withColumn("Revenue_Band",
    F.when(F.col("Transaction_Amount") >= 50000, "High")
     .when(F.col("Transaction_Amount") >= 10000, "Medium")

     .otherwise("Low"))


df_enriched.select("Customer_ID", "Transaction_Amount", "Revenue_Band").show(3)

+-----------+------------------+------------+
|Customer_ID|Transaction_Amount|Revenue_Band|
+-----------+------------------+------------+
|CUST003    |5752.36           |Low         |
|CUST009    |28959.12          |Medium      |
|CUST010    |72098.18          |High        |
+-----------+------------------+------------+

데이터 품질 검사

total_rows = df_enriched.count()
distinct_rows = df_enriched.distinct().count()


null_rate = (
    df_enriched.filter(F.col("Customer_ID").isNull())
    .count() / total_rows * 100
)


print(f"Total rows:   {total_rows}")
print(f"Duplicates:   {total_rows - distinct_rows}")
print(f"Null rate:    {null_rate:.2f}%")

Total rows:   33,223
Duplicates:   0
Null rate:    0.00%

연습해 봅시다!

Databricks에서 Spark SQL로 데이터 변환하기