Apache PySpark：如何每日对数千台设备进行批处理分析

针对你纠结的两种方案，直接对比优缺点，再给出更贴合场景的最优实现思路：

一、单作业循环处理（你当前的代码思路） #

优点 #

• 资源利用率高：只启动一次Spark集群，避免多次初始化SparkSession的冗余开销
• 运维成本低：只需提交一个作业，监控、调度都更省心
• 配置复用：依赖包、集群配置只加载一次，节省启动时间

缺点 #

• 串行处理效率低：你的代码是循环逐个读取文件，相当于单线程跑，完全浪费了Spark的分布式并行能力
• 单点故障风险：某个设备处理失败会直接中断整个作业（需要额外加异常捕获逻辑兜底）
• 资源无隔离：如果某台设备数据量极大，会占用所有executor资源，拖慢其他设备的处理进度

二、每个设备单独提交作业 #

优点 #

• 故障隔离性好：单个设备处理失败不会影响其他作业的运行
• 资源弹性可控：可以为不同数据量的设备单独分配资源（比如大设备用多核心，小设备用少资源）
• 调度灵活：高优先级设备可以优先执行

缺点 #

• 资源浪费严重：每个作业都要启动Spark上下文，重复加载依赖、初始化集群，额外开销极大
• 运维复杂度高：要管理几十甚至上百个作业的提交、日志、监控，成本直线上升
• 集群调度压力大：大量作业同时提交会导致资源排队，反而拖慢整体处理速度

三、推荐方案：单作业批量并行处理（优化你的代码） #

结合IoT设备数据格式统一的特点，改成批量读取所有文件，利用Spark的分布式能力按设备并行处理，兼顾效率和运维便利性。

优化后的代码示例：

from pyspark.sql import SparkSession
from pyspark.sql.functions import input_file_name, regexp_extract, avg, max, count

# 初始化SparkSession，配置MongoDB连接
spark = SparkSession.builder \
    .appName("BatteryDataAnalysis_Batch") \
    .config("spark.mongodb.output.uri", "mongodb://localhost:27017/iot_db.battery_results") \
    .getOrCreate()

# 批量读取HDFS上所有设备的CSV文件（用通配符一次性加载）
data_source_path = "hdfs://your-hdfs-path/data/battery_data/*.csv"
df = spark.read.csv(data_source_path, header=True, inferSchema=True)

# 从文件名提取设备ID（假设文件名格式为 device_123.csv，可根据实际格式调整正则）
df = df.withColumn("device_id", regexp_extract(input_file_name(), r"device_(\d+)\.csv", 1))

# 按设备ID并行处理，这里以统计每个设备的电池指标为例
device_analysis_df = df.groupBy("device_id").agg(
    avg("voltage").alias("avg_voltage"),
    max("temperature").alias("max_temperature"),
    count("*").alias("total_data_points")
)

# 将结果批量写入MongoDB，支持自动分批次写入
device_analysis_df.write.format("mongo").mode("overwrite").save()

spark.stop()

额外优化建议 #

• 异常处理：如果需要保证单个设备失败不影响全局，可以用foreachPartition遍历每个设备分区，在分区内加try-except逻辑
• 增量处理：如果是每日新增数据，建议按日期对HDFS数据分区（比如/data/battery_data/date=20240520/），读取时指定日期范围，减少数据扫描量
• MongoDB写入优化：调整spark.mongodb.output.batch.size参数，增大批量写入的批次，减少数据库连接次数
• 资源配置：根据总数据量设置合适的executor数量、内存，比如--executor-memory 4G --num-executors 8，避免内存溢出

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