如何在PySpark Streaming中按地点与车辆类别计算平均速度？

需求说明 #

现有交通计数器输出的CSV格式DStream数据，每条记录包含地点、车辆类型、速度信息，需要计算每个地点下不同车辆类别的平均速度，输出指定格式的表格结果。

示例输入数据 #

Location,Vehicle,Speed,
tracker1,car,57,
tracker1,car,90,
tracker1,mbike,81,
tracker1,mbike,65,
tracker2,car,69,
tracker2,car,34,
tracker2,mbike,29,
tracker2,mbike,76,

期望输出格式 #

Location |  Car | MBike
Tracker 1| 73.5 |  73.0
Tracker 2| 51.5 |  52.5

解决方案（Scala版本） #

import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._

// 初始化StreamingContext，批处理间隔设为10秒
val ssc = new StreamingContext(spark.sparkContext, Seconds(10))

// 从Socket获取CSV流（实际场景可替换为文件/Kafka等数据源）
val csvDStream = ssc.socketTextStream("localhost", 9999)

// 解析并格式化数据：过滤表头，处理多余逗号，转换字段格式
val parsedDStream = csvDStream
  .filter(!_.startsWith("Location"))
  .map(line => {
    val parts = line.split(",").filter(_.nonEmpty)
    // 将trackerX格式化为Tracker X
    val location = parts(0) match {
      case s"tracker$num" => s"Tracker $num"
      case _ => parts(0)
    }
    val vehicle = parts(1).capitalize // 转成Car/MBike
    val speed = parts(2).toDouble
    ((location, vehicle), (speed, 1L)) // 键：(地点,车辆类型)，值：(速度,计数)
  })

// 按键聚合，计算每个分组的总速度和总记录数
val sumCountDStream = parsedDStream.reduceByKey((a, b) => (a._1 + b._1, a._2 + b._2))

// 计算平均速度
val avgSpeedDStream = sumCountDStream.mapValues { case (totalSpeed, count) => totalSpeed / count }

// 按地点合并不同车辆类型的平均速度
val resultDStream = avgSpeedDStream
  .map { case ((location, vehicle), avg) => (location, Map(vehicle -> avg)) }
  .reduceByKey((map1, map2) => map1 ++ map2)

// 格式化输出为指定表格样式
resultDStream.foreachRDD { rdd =>
  if (!rdd.isEmpty()) {
    println("Location |  Car | MBike")
    rdd.collect().foreach { case (location, avgMap) =>
      val carAvg = avgMap.getOrElse("Car", 0.0)
      val mbikeAvg = avgMap.getOrElse("MBike", 0.0)
      println(f"$location%-9s| $carAvg%5.1f | $mbikeAvg%5.1f")
    }
  }
}

// 启动StreamingContext并等待终止
ssc.start()
ssc.awaitTermination()

解决方案（Python版本） #

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# 初始化StreamingContext，批处理间隔10秒
sc = SparkContext("local[2]", "TrafficSpeedAvg")
ssc = StreamingContext(sc, 10)

# 从Socket获取CSV流（可替换为实际数据源）
csv_dstream = ssc.socketTextStream("localhost", 9999)

def parse_line(line):
    parts = [p.strip() for p in line.split(",") if p.strip()]
    # 过滤表头
    if parts[0] == "Location":
        return None
    # 格式化地点名称
    if parts[0].startswith("tracker"):
        location = f"Tracker {parts[0][7:]}"
    else:
        location = parts[0]
    vehicle = parts[1].capitalize()
    speed = float(parts[2])
    return ((location, vehicle), (speed, 1))

# 解析数据并过滤无效记录
parsed_dstream = csv_dstream.map(parse_line).filter(lambda x: x is not None)

# 聚合总速度和记录数
sum_count_dstream = parsed_dstream.reduceByKey(lambda a, b: (a[0]+b[0], a[1]+b[1]))

# 计算平均速度
avg_speed_dstream = sum_count_dstream.mapValues(lambda x: x[0]/x[1])

# 按地点合并车辆类型的平均速度
result_dstream = avg_speed_dstream.map(lambda x: (x[0][0], {x[0][1]: x[1]}))\
    .reduceByKey(lambda a, b: {**a, **b})

# 格式化输出表格
def print_table(rdd):
    if not rdd.isEmpty():
        print("Location |  Car | MBike")
        for location, avg_map in rdd.collect():
            car_avg = avg_map.get("Car", 0.0)
            mbike_avg = avg_map.get("MBike", 0.0)
            print(f"{location:<9}| {car_avg:5.1f} | {mbike_avg:5.1f}")

result_dstream.foreachRDD(print_table)

# 启动流处理并等待终止
ssc.start()
ssc.awaitTermination()

关键说明 #

1. 数据解析：处理CSV末尾的多余逗号，通过过滤空字符串避免解析错误，同时格式化地点和车辆类型名称以匹配期望输出。
2. 聚合逻辑：先以(地点,车辆类型)为键聚合总速度和记录数，计算平均值后再按地点合并，确保每个地点的所有车辆类型平均速度在同一条结果中。
3. 数据源适配：示例使用Socket流，实际可根据需求替换为文件流、Kafka、Flume等Spark Streaming支持的输入源。

内容的提问来源于stack exchange，提问作者Muhammad Allain