1. 出租车轨迹数据清洗实战指南第一次处理出租车GPS数据时我被原始数据里各种乱码符号惊呆了——$、这些字符像打地鼠一样随机出现在各个字段。后来发现这类脏数据在真实业务场景中太常见了比如传感器异常、系统对接问题都会导致数据污染。今天我就用Spark SQL手把手教你搞定这个难题。先看个典型脏数据样本TRIP_ID|CALL_TYPE|TAXI_ID $1372636858|B|20000589$ 1372637303|C|20000596这类数据直接分析会报错我们需要两个核心技能用SparkSQL正确读取含特殊分隔符的CSV编写UDF函数清洗不规则字符读取数据时这几个参数最关键df spark.read.option(header, True) \ .option(delimiter, |) \ # 指定竖线分隔符 .option(inferSchema, True) \ .csv(/data/taxi_trips.csv)遇到tab分隔符(\t)的情况要特别注意建议先用print(df.first()) # 检查首行数据是否解析正确2. 自定义UDF开发详解Spark的UDF(User Defined Function)就像SQL里的瑞士军刀能扩展各种数据处理能力。我们案例需要清洗两类特殊字符from pyspark.sql.functions import udf from pyspark.sql.types import StringType def clean_special_chars(text): import re return re.sub(r[\$\], , str(text)) # 注册UDF clean_udf udf(clean_special_chars, StringType()) # 使用示例 df_clean df.withColumn(TRIP_ID, clean_udf(df[TRIP_ID]))正则表达式进阶技巧r[\$\]表示匹配$或出现1次或多次如果想保留特定符号比如#修改为r[^a-zA-Z0-9#]处理中文时建议加上unicode范围r[^\u4e00-\u9fa5]实际项目中我推荐这种链式处理clean_udf udf(lambda x: str(x) .replace($,) .replace(,) .strip(), StringType())3. 工业级数据清洗方案真实场景不能只处理特殊字符还需要全套清洗流程3.1 字段校验规则字段校验规则处理方式TRIP_ID长度18位数字自动补零/截断TIMESTAMP10位Unix时间戳无效值置为NULLPOLYLINEJSON格式坐标串语法校验3.2 完整清洗代码# 初始化Spark spark SparkSession.builder \ .appName(TaxiDataCleaning) \ .config(spark.sql.shuffle.partitions, 200) \ .getOrCreate() # 多级清洗UDF def full_clean(text, field_type): text str(text) # 第一阶段基础清洗 text re.sub(r[^\w\.-], , text) # 第二阶段类型专项处理 if field_type timestamp: if not text.isdigit(): return None elif field_type id: text text.zfill(18)[:18] return text # 注册多参数UDF schema StructType([ StructField(value, StringType()), StructField(type, StringType()) ]) full_clean_udf udf(full_clean, StringType(), schema) # 执行清洗 df_clean df.withColumn(TRIP_ID, full_clean_udf(df[TRIP_ID], lit(id)))4. 性能优化与异常处理在大数据量下(我处理过10TB的轨迹数据)这几个优化点很关键分区策略按日期分区处理df.repartition(100, date_col)缓存中间结果df_clean.cache() print(df_clean.count()) # 触发缓存错误处理机制from pyspark.sql.functions import when df_clean df.withColumn(TRIP_ID, when(df[TRIP_ID].rlike(^\d$), df[TRIP_ID]) .otherwise(None))常见踩坑点直接dropna()会丢失大量数据建议先分析缺失模式复杂正则可能导致性能瓶颈先用sample测试本地测试OK的UDF可能在集群报错注意依赖库问题轨迹数据清洗只是第一步清洗后的数据可以用来做热点区域分析路径规划优化异常驾驶行为检测记得每次处理前先做数据采样检查df.sample(0.01).show(5)