1. 从零开始搭建Spark数据分析环境

第一次接触Spark时，我被它复杂的配置过程折磨得不轻。后来发现用Docker容器化部署能省去80%的麻烦，这里分享我的避坑指南。你需要先安装好Docker Desktop，然后执行下面这条魔法般的命令：

docker run -p 4040:4040 -p 8080:8080 --name spark-demo bitnami/spark:latest

这个镜像已经预装了Spark 3.3和所有依赖项，启动后访问localhost:4040就能看到Spark UI。我建议同时安装JupyterLab作为Python开发环境：

# 在容器内安装Jupyter
!pip install jupyterlab pyspark pandas

实测发现，处理出租车轨迹这种地理数据时，geopandas库必不可少。但要注意版本兼容性问题 - Spark 3.x最好搭配Python 3.8+。曾经因为版本冲突浪费了我整个下午，后来固定使用这个组合再没出过问题。

2. 出租车数据清洗的五个关键步骤

拿到原始GPS数据时，我常看到这样的混乱场景：缺失的车辆ID、乱码的经纬度、错误的时间戳。通过多次实战总结了这套清洗流程：

2.1 处理异常坐标点

出租车GPS常出现"跳点"现象，比如相邻两点距离突然超过1公里。我用这个PySpark函数自动过滤：

from pyspark.sql.functions import udf
from pyspark.sql.types import BooleanType

@udf(BooleanType())
def is_valid_point(prev_lon, prev_lat, curr_lon, curr_lat):
    # 两点间距离超过1公里视为异常
    return haversine(prev_lon, prev_lat, curr_lon, curr_lat) < 1000

2.2 时间戳标准化

不同设备的时间格式五花八门，这个转换模板能处理90%的情况：

SELECT 
  TRIP_ID,
  CASE 
    WHEN TIMESTAMP RLIKE '^\\d{10}$' THEN from_unixtime(CAST(TIMESTAMP AS LONG))
    WHEN TIMESTAMP RLIKE '^\\d{4}-' THEN to_timestamp(TIMESTAMP)
    ELSE NULL
  END AS STANDARD_TIME
FROM raw_data

3. 轨迹特征工程实战技巧

3.1 计算实时行驶速度

通过相邻坐标点的时间差和距离，可以估算瞬时速度。这里有个优化点：使用窗口函数比自连接快10倍：

from pyspark.sql.window import Window

window_spec = Window.partitionBy("TAXI_ID").orderBy("TIMESTAMP")

df = df.withColumn("prev_lat", lag("latitude").over(window_spec)) \
       .withColumn("prev_lon", lag("longitude").over(window_spec)) \
       .withColumn("prev_time", lag("TIMESTAMP").over(window_spec))

3.2 识别停靠点

出租车等客时的坐标聚集特征很明显。我用DBSCAN算法检测，但需要先把数据采样到本地：

from sklearn.cluster import DBSCAN

# 每辆车抽取500个点作为样本
sample_points = df.groupBy("TAXI_ID") \
                 .agg(collect_list("latitude").alias("lats"), 
                      collect_list("longitude").alias("lons")) \
                 .toPandas()

4. 可视化分析的隐藏技巧

Matplotlib默认渲染10万+轨迹点会卡死，我找到两个解决方案：

方案一：使用Datashader

import datashader as ds
from datashader import transfer_functions as tf

canvas = ds.Canvas(plot_width=800, plot_height=600)
agg = canvas.points(df, 'longitude', 'latitude')
tf.shade(agg, cmap=["blue", "red"])

方案二：PySpark原生近似计算

df.sample(fraction=0.01).toPandas().plot.scatter(x='longitude', y='latitude')

曾经为了展示全市出租车热力图，我不得不租用AWS的r5.4xlarge机型。后来改用上述方法，在笔记本上就能流畅运行。

5. 性能优化实战记录

5.1 分区策略选择

按车辆ID分区是最初方案，但发现数据倾斜严重。后来改用空间网格分区：

df = df.withColumn("grid_id", 
    (floor(col("longitude")/0.01)*100 + floor(col("latitude")/0.01)))

5.2 内存缓存技巧

多次使用的中间结果一定要缓存，但要注意存储级别：

from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK_SER)  # 序列化节省空间

有次忘记释放缓存导致集群内存爆满，现在养成了在finally块清理的习惯：

try:
    df.cache()
    # 处理逻辑...
finally:
    df.unpersist()

6. 典型业务场景解决方案

6.1 高峰时段识别

通过时间分布直方图发现早高峰比晚高峰更集中：

df.withColumn("hour", hour("TIMESTAMP")) \
  .groupBy("hour") \
  .count() \
  .orderBy("hour") \
  .toPandas() \
  .plot.bar(x="hour", y="count")

6.2 热门路线挖掘

先用Geohash压缩空间维度，再找高频组合：

SELECT 
  geohash(start_lat, start_lon, 6) as start_geohash,
  geohash(end_lat, end_lon, 6) as end_geohash,
  COUNT(*) as trip_count
FROM trips
GROUP BY 1, 2
ORDER BY 3 DESC
LIMIT 10

7. 生产环境部署经验

7.1 参数调优配置

这些spark-submit参数让我的作业速度提升3倍：

spark-submit \
  --executor-memory 8G \
  --driver-memory 4G \
  --conf spark.sql.shuffle.partitions=200 \
  --conf spark.default.parallelism=200 \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  your_script.py

7.2 监控指标关注点

在Spark UI中我主要看这几个指标：

• 每个stage的GC时间（超过10%就要优化）
• 数据倾斜度（最大task时长/平均时长 >3就是倾斜）
• 存储内存占比（低于70%说明缓存不够）

曾经因为没监控shuffle读写量，导致作业卡住3小时。现在养成了在代码里埋点记录的习惯：

df.rdd.countApprox(timeout=1000)  # 强制触发执行计划