การทำความสะอาดข้อมูลและการตรวจสอบคุณภาพ

การแปลงข้อมูลด้วย Spark SQL ใน Databricks

Disha Mukherjee

Lead Data Engineer

3 ปัญหาในข้อมูลจริง

recraft: half: กองเอกสารและไฟล์ที่กระจัดกระจาย มีข้อผิดพลาดและช่องว่าง บนพื้นหลังโปร่งใส

ค่าที่หายไป ในคอลัมน์สำคัญ
แถวซ้ำ จากการนำเข้าข้อมูลซ้ำ
ระเบียนที่ไม่สมเหตุสมผล

ทำไมต้องกำหนด schema อย่างชัดเจน?

recraft: half: นักวิทยาศาสตร์ที่งุนงง มีเครื่องหมายคำถามลอยอยู่เหนือแถวในสเปรดชีต บนพื้นหลังโปร่งใส

Spark อาจอ่านประเภทคอลัมน์ผิดพลาดได้
คอลัมน์ตัวเลขอาจถูกอนุมานเป็น string
StructType กำหนดประเภทข้อมูลอย่างชัดเจนทุกครั้งที่รัน

การ import

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


from pyspark.sql import functions as F

การกำหนด schema

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

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      |
+-----------+------------------+--------+--------+

การลบและเติมค่า null

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%

มาฝึกกันเถอะ!

การแปลงข้อมูลด้วย Spark SQL ใน Databricks