1. 电影票房数据清洗的核心挑战

电影票房数据清洗是数据分析过程中最关键的预处理环节之一。在实际项目中，原始数据往往存在各种"脏数据"问题，比如特殊字符、单位不统一、日期格式混乱等。以我们常见的电影票房数据为例，一个看似简单的"上映天数"字段就可能包含"上映25天"、"展映"、"点映"等多种表达方式。

我曾经处理过一个真实项目的数据集，其中"当日综合票房"字段竟然有"1.5亿"、"162.4万"、"75.1"三种不同的表示方式。这种数据如果不经过统一处理，后续的分析计算根本无法进行。更棘手的是，当使用float或double类型处理货币单位转换时，经常会出现精度丢失问题，比如1.14亿转换为万时，可能会得到11399.999999999998这样的结果。

2. 搭建Spark开发环境

在开始数据清洗前，我们需要准备好Spark开发环境。这里我推荐使用Spark 3.x版本，它对SQL功能和性能都有显著优化。对于本地开发测试，可以这样初始化SparkSession：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MovieBoxOfficeCleaning") \
    .config("spark.sql.shuffle.partitions", "4") \
    .getOrCreate()

如果是处理大规模数据，建议在集群环境下运行，并适当调整内存配置：

spark = SparkSession.builder \
    .appName("MovieBoxOfficeCleaning") \
    .config("spark.executor.memory", "8g") \
    .config("spark.driver.memory", "4g") \
    .getOrCreate()

3. 数据加载与初步探索

加载数据是第一步，但很多新手会忽略数据探索的重要性。我建议先用Spark的show()和printSchema()方法快速了解数据概况：

movies_df = spark.read.csv("movies.csv", header=True, inferSchema=True)
movies_df.show(5, truncate=False)
movies_df.printSchema()

通过初步探索，我们可能会发现几个常见问题：

1. 数值字段中混有"万"、"亿"等单位
2. 日期字段格式不一致
3. 存在需要过滤的特殊放映类型（如"点映"、"展映"）
4. 某些字段存在空值

4. 处理特殊放映类型数据

"上映天数"字段中的特殊值需要特别注意。根据业务需求，我们通常需要过滤掉"零点场"、"点映"等特殊放映类型的数据。在Spark中，可以使用filter()结合正则表达式实现：

from pyspark.sql.functions import col

# 过滤特殊放映类型
cleaned_df = movies_df.filter(
    (~col("movie_days").contains("零点场")) &
    (~col("movie_days").contains("点映")) &
    (~col("movie_days").contains("展映")) &
    (~col("movie_days").contains("重映"))
)

这里有个实用技巧：在过滤前先统计各类特殊值的数量，可以帮助我们评估数据质量：

special_types = ["零点场", "点映", "展映", "重映"]
for t in special_types:
    count = movies_df.filter(col("movie_days").contains(t)).count()
    print(f"包含'{t}'的记录数: {count}")

5. 日期转换的实用技巧

日期处理是数据清洗中最容易出错的环节之一。我们需要将"上映天数"转换为具体的"上映日期"。这里有几个关键点需要注意：

1. "上映首日"应直接使用"当前日期"
2. 空值应标记为"往期电影"
3. 正常的天数需要计算具体日期

在Spark中，我们可以使用UDF（用户自定义函数）来处理这种复杂逻辑：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
from datetime import datetime, timedelta
import re

def calculate_release_date(movie_days, current_time):
    if not movie_days:
        return "往期电影"
    if movie_days == "上映首日":
        return current_time
    if "天" in movie_days:
        days = int(re.sub("[^0-9]", "", movie_days))
        release_date = datetime.strptime(current_time, "%Y-%m-%d") - timedelta(days=days-1)
        return release_date.strftime("%Y-%m-%d")
    return "往期电影"

date_udf = udf(calculate_release_date, StringType())
cleaned_df = cleaned_df.withColumn("releaseDate", date_udf(col("movie_days"), col("current_time")))

6. 票房数据的精确处理

货币单位的转换需要特别注意精度问题。我强烈建议使用Decimal类型而不是float或double来处理金融数据。以下是处理"当日综合票房"和"当前总票房"字段的完整方案：

from pyspark.sql.functions import udf
from pyspark.sql.types import DecimalType
from decimal import Decimal
import re

def convert_box_office(value):
    if not value:
        return None
    value = value.replace(",", "")
    if "亿" in value:
        num = Decimal(re.sub("[^0-9.]", "", value))
        return num * Decimal("10000")
    elif "万" in value:
        return Decimal(re.sub("[^0-9.]", "", value))
    else:
        return Decimal(value)

box_office_udf = udf(convert_box_office, DecimalType(20, 2))
cleaned_df = cleaned_df.withColumn("boxoffice", box_office_udf(col("boxoffice"))) \
                      .withColumn("total_boxoffice", box_office_udf(col("total_boxoffice")))

7. 数据保存的最佳实践

清洗后的数据保存也有讲究。根据数据量大小，我们需要考虑分区策略和文件格式。对于中小规模数据，建议：

# 重新分区并保存为CSV
cleaned_df.coalesce(1) \
         .write \
         .option("delimiter", "\t") \
         .option("header", "true") \
         .mode("overwrite") \
         .csv("output/movies_cleaned")

如果数据量很大（超过GB级别），可以考虑使用Parquet格式，并适当增加分区数：

cleaned_df.repartition(8) \
          .write \
          .mode("overwrite") \
          .parquet("output/movies_cleaned_parquet")

8. 常见问题与调试技巧

在实际项目中，我遇到过几个典型问题值得分享：

1. 日期计算错误：曾经因为忽略了"上映天数"需要减1的逻辑（首日算第1天），导致所有日期都提前了一天。建议用少量测试数据验证计算逻辑。

2. 性能问题：当处理大规模数据时，不当的UDF使用会导致性能下降。可以尝试：

   • 优先使用内置函数
   • 对UDF进行优化
   • 适当增加executor内存

3. 字符编码问题：特别是处理中文数据时，确保读取和写入使用一致的编码（通常UTF-8）。

调试时可以充分利用Spark UI（默认4040端口）来监控作业执行情况，找出性能瓶颈。对于数据验证，我习惯抽样检查：

# 抽样检查日期转换结果
cleaned_df.select("movie_days", "current_time", "releaseDate") \
          .sample(0.1) \
          .show(20, truncate=False)

9. 扩展应用与优化建议

掌握了基础的数据清洗技巧后，我们可以进一步优化流程：

1. 自动化测试：为数据清洗逻辑编写单元测试，特别是边界条件（如空值、特殊字符等）。

2. 参数化配置：将过滤条件、字段映射等提取到配置文件中，提高代码复用性。

3. 监控与告警：对数据质量建立监控指标（如空值率、异常值比例等），设置阈值告警。

4. 性能优化：对于定期执行的清洗任务，可以考虑：

   • 缓存常用数据集
   • 优化shuffle操作
   • 使用广播变量

# 性能优化示例
cleaned_df.cache()  # 缓存清洗后的数据
broadcast_rules = spark.sparkContext.broadcast(cleaning_rules)  # 广播清洗规则

10. 真实案例经验分享

在最近一个电影数据分析项目中，我们遇到了一个有趣的问题：某些电影的"上映天数"超过了实际可能的范围（比如"上映500天"但数据集时间跨度只有一年）。经过调查发现，这些是经典电影的重映记录。针对这种情况，我们采取了以下处理方案：

1. 保留重映记录但标记为特殊类别
2. 为原始上映和重映分别计算日期
3. 在分析时区分不同放映类型

这个案例让我深刻体会到，数据清洗不仅是技术问题，更需要理解业务背景。有时候"脏数据"反而包含了重要的业务信息。