Spark海量心率数据聚合实战：从数据倾斜到性能调优

1. 从一个心率日志说起：这个案例到底在算什么

手头有一批可穿戴设备导出的心率日志，格式大概是“用户ID、时间戳、心率值”三列，一天下来几百万行，一个中型应用攒上几个月就是几十亿行。业务方想知道：每个用户在这段时间里的平均心率是多少？听起来简单到用Excel都能做，但数据量一上来，单机脚本直接趴窝。这个案例要解决的就是这个问题——用Spark把海量心率记录按用户聚合，算出平均心率，并且保证在集群上跑得又快又稳。

适合谁看？如果你刚接触Spark，会写一点Scala或Python，但一遇到“数据倾斜”“分区”“shuffle”这些词就发怵，那这个案例正好拿来练手。它不涉及机器学习，也不搞图计算，就是最朴素的“读数据、分组、聚合、写结果”四步走。但恰恰是这种最基础的聚合，藏着Spark调优里最典型的坑：数据倾斜、分区数设置、序列化开销、空值处理。把这一套吃透，后面再遇到求和、计数、去重、TopN，思路都是通的。

我拿到的原始数据是CSV格式，每行三个字段：user_id、timestamp、heart_rate。心率值范围在40到200之间，偶尔有0或者负数，那是设备佩戴不紧导致的无效读数。时间戳是毫秒级，但算平均心率其实用不上时间维度，除非你要按小时或按天分段统计。这里先做全局平均，后面再扩展成滑动窗口。

提示：心率数据属于个人健康信息，实际生产中必须做脱敏处理，比如把user_id哈希成不可逆的字符串，或者直接在聚合后丢弃原始ID。案例里为了演示方便，用的都是模拟生成的假ID。

整个项目的技术栈很轻：Spark 3.x（2021年那会儿2.4和3.0并存，现在建议直接上3.x）、HDFS或者本地文件系统做存储、Scala 2.12或者Python 3.7以上。集群规模不用大，单机伪分布式就能跑通逻辑，但要想看到调优效果，至少得有个三四台机器的Standalone或者YARN集群。我自己的测试环境是一台16核32G的虚拟机跑Spark Local模式，外加一个三节点的小集群做对比。

2. 为什么选Spark而不是Pandas或SQL

2.1 数据量决定工具边界

先算一笔账。假设有1000万用户，每人每天产生1440条心率记录（每分钟一次），一天就是144亿行。每行按最简单的二进制格式存，user_id用Long（8字节）、timestamp用Long（8字节）、heart_rate用Int（4字节），加起来20字节，一天的数据量就是288GB。这还没算上副本和元数据。Pandas读这种量级的数据，内存直接爆掉，就算用分块读取，groupby的时候也会因为哈希表太大而慢到无法接受。

传统关系型数据库呢？单表288GB，按user_id做索引，查询“每个用户的平均心率”需要全表扫描加分组聚合。MySQL在千万级数据上跑这种查询，没个几分钟下不来，而且并发一高就锁表。Spark的优势在于它把数据切成分区，每个分区独立计算局部平均，最后再合并。这就是MapReduce式的分而治之，线性扩展，加机器就能提速。

2.2 Spark的DataFrame API比RDD更省心

早期Spark教程喜欢用RDD的map和reduceByKey，代码写起来灵活但容易踩坑。比如下面这种写法：

val rdd = sc.textFile("hdfs://path/to/heart_rate.csv") val avgRdd = rdd.map { line => val fields = line.split(",") (fields(0), (fields(2).toInt, 1)) }.reduceByKey { case ((sum1, cnt1), (sum2, cnt2)) => (sum1 + sum2, cnt1 + cnt2) }.mapValues { case (sum, cnt) => sum.toDouble / cnt }

这段代码逻辑没错，但问题不少：split和toInt没有异常处理，遇到脏数据直接抛异常；reduceByKey默认分区数继承父RDD，如果父RDD分区太少，每个分区要处理的数据量巨大，容易OOM；而且RDD没有schema，字段顺序全靠约定，后期维护容易搞混。

换成DataFrame API就清爽多了：

val df = spark.read .option("header", "true") .option("inferSchema", "true") .csv("hdfs://path/to/heart_rate.csv") val avgDf = df.filter($"heart_rate" > 0 && $"heart_rate" < 250) .groupBy("user_id") .agg(avg("heart_rate").as("avg_hr"))

filter自动跳过无效值，groupBy和agg由Catalyst优化器生成物理计划，底层自动做部分聚合（partial aggregation），减少shuffle数据量。而且DataFrame有schema，字段类型明确，写错了编译期就能发现。2021年那会儿DataFrame已经非常成熟，除非你要做非常底层的自定义分区，否则没必要用RDD。

2.3 和Flink、Presto的简单对比

有人会问：实时心率检测不是Flink更合适吗？没错，如果是要做秒级告警，比如心率超过180立刻通知，那Flink的流处理是首选。但这个案例的需求是“算历史数据的平均心率”，属于批处理场景，Spark的吞吐量和生态成熟度更占优。Presto（现在叫Trino）适合交互式查询，但它的聚合依赖内存，数据量一大就容易OOM，而且没有Spark那种磁盘溢写机制。所以选Spark是权衡了数据量、延迟要求和运维成本后的结果。

3. 数据读取与清洗：脏数据比算法更棘手

3.1 心率数据的典型脏数据形态

真实设备导出的心率日志，脏起来五花八门。我整理了几种最常见的：

处理这些脏数据，我习惯在读取阶段就用filter一次性搞定，而不是等到聚合时再判断。因为filter是窄依赖，不会触发shuffle，越早过滤掉无效数据，后续shuffle的数据量越小。

val cleanDf = rawDf .filter($"user_id".isNotNull && $"heart_rate".isNotNull) .filter($"heart_rate" > 30 && $"heart_rate" < 220) .filter($"timestamp" > 0) .dropDuplicates("user_id", "timestamp")

dropDuplicates会触发一次shuffle，因为要去重就必须把相同key的数据拉到同一个分区。如果数据量特别大，这一步开销不小。我的经验是：如果重复率不高（比如低于1%），可以不做去重，直接聚合，因为重复记录对平均值的影响微乎其微。但如果重复率超过5%，那就必须去重，否则平均值会偏向重复次数多的那个值。

3.2 用累加器统计脏数据比例

光过滤还不够，你得知道过滤掉了多少数据，否则心里没底。Spark的累加器（Accumulator）就是干这个的：

val invalidCount = spark.sparkContext.longAccumulator("invalid_hr") val cleanDf = rawDf.filter { row => val hr = row.getAs[Int]("heart_rate") if (hr <= 30 || hr >= 220) { invalidCount.add(1) false } else true } // 触发action后打印 println(s"过滤掉无效心率记录：${invalidCount.value}条")

注意累加器只能在action操作里更新，而且如果任务重试，累加器可能会重复计数。所以它适合做监控指标，不适合做精确的财务计算。

3.3 分区数设置：别让默认值坑了你

spark.read.csv读进来的DataFrame，分区数取决于spark.default.parallelism和文件大小。如果是HDFS上的一个大文件，默认按128MB一个block切分，每个block一个分区。假设文件是10GB，那就是80个分区。80个分区对于聚合来说太少了，每个分区要处理128MB数据，groupBy的时候每个分区内部要维护一个巨大的哈希表，内存压力很大。

我的做法是：读取后立刻repartition到合适的分区数。经验公式是：分区数 = 数据量(MB) / 128，但不少于集群总核数的2到3倍。比如10GB数据，80个分区，集群有30个核，那可以repartition到90到120个分区。

val repartitionedDf = cleanDf.repartition(120, $"user_id")

按user_id做repartition的好处是，相同用户的数据会落到同一个分区，后续groupBy时不需要再shuffle。但坏处是如果某些用户的数据特别多（比如VIP用户心率记录频率更高），会导致数据倾斜。这个后面再细说。

4. 聚合计算：avg()背后的分布式逻辑

4.1 部分聚合与最终聚合

groupBy("user_id").agg(avg("heart_rate"))看起来是一步操作，实际上Spark在底层拆成了两步：部分聚合（partial aggregation）和最终聚合（final aggregation）。部分聚合在每个分区内部先算局部平均，输出(user_id, partial_sum, partial_count)；最终聚合把相同user_id的partial结果拉到一起，用total_sum / total_count算出全局平均。

这个设计的好处是减少了shuffle数据量。假设有1亿条记录，1000万用户，每个用户10条记录。如果不做部分聚合，shuffle要传输1亿条(user_id, heart_rate)；做了部分聚合后，每个分区先合并，假设有100个分区，每个分区输出1000万/100=10万条(user_id, sum, count)，总共1000万条，数据量降到十分之一。

你可以通过explain()查看物理计划，确认是否有PartialAggregate：

avgDf.explain(true)

输出里如果看到HashAggregate(keys=[user_id], functions=[partial_avg(heart_rate)])，说明部分聚合生效了。如果只有HashAggregate(keys=[user_id], functions=[avg(heart_rate)])，那说明优化器没做部分聚合，可能是数据类型不支持或者配置有问题。

4.2 数据倾斜：某个用户的心率记录特别多

数据倾斜是聚合操作的头号杀手。假设有个用户是马拉松运动员，设备每秒记录一次心率，一天就是86400条，而普通用户每分钟一次，一天1440条。两者相差60倍。groupBy的时候，这个运动员的数据全部分到一个分区，那个分区的任务要处理86400条，其他分区只处理1440条，整个作业的耗时取决于最慢的那个任务。

怎么发现倾斜？看Spark UI的Stage页面，如果某个Task的Duration远大于其他Task，或者Shuffle Read Size特别大，那就是倾斜了。我遇到过最夸张的一次，一个Task处理了2GB数据，其他Task只有20MB，跑了40分钟还没结束。

解决倾斜有几种思路：

加盐打散：给user_id拼接一个随机前缀，比如user_id + "_" + (0 until 10).random，这样同一个用户的数据被分到10个分区，聚合后再去掉前缀做二次聚合。缺点是代码复杂，而且二次聚合又要一次shuffle。

val saltedDf = cleanDf.withColumn("salted_id", concat($"user_id", lit("_"), (rand() * 10).cast("int"))) val partialAvg = saltedDf.groupBy("salted_id") .agg(sum("heart_rate").as("sum_hr"), count("heart_rate").as("cnt")) val finalAvg = partialAvg .withColumn("user_id", split($"salted_id", "_")(0)) .groupBy("user_id") .agg(sum("sum_hr") / sum("cnt"))

广播小表：如果倾斜的user_id是已知的少数几个，可以把它们单独拿出来处理，剩下的用正常逻辑。比如先统计每个用户的记录数，把超过阈值的用户ID收集到Driver端，然后分两个DataFrame处理，最后union。

调整分区数：增加分区数可以让数据更均匀，但治标不治本。如果某个key的数据量是其他key的100倍，分区数再多也没用，因为同一个key只能在一个分区里。

我的建议是：先看倾斜程度。如果最慢Task和平均Task的耗时差距在2倍以内，可以忽略；3到5倍，考虑加盐；超过10倍，必须专门处理。

4.3 空值处理：avg()遇到null会怎样

Spark的avg()函数会自动忽略null值。也就是说，如果某个用户的所有心率记录都是null，那他的平均心率返回null。这通常不是我们想要的，业务方可能希望显示“无数据”或者0。可以在agg之后用na.fill(0)填充，或者用coalesce(avg("heart_rate"), lit(0))。

但要注意：如果null是因为设备故障导致的，填0会拉低整体平均值，造成误导。更合理的做法是单独统计有效记录数，如果有效记录数低于某个阈值（比如10条），就不输出这个用户的平均心率。

val resultDf = cleanDf.groupBy("user_id") .agg( avg("heart_rate").as("avg_hr"), count("heart_rate").as("valid_cnt") ) .filter($"valid_cnt" >= 10)

5. 性能调优：从30分钟到3分钟的实操记录

5.1 序列化：Kryo比Java序列化快10倍

Spark默认用Java序列化，慢且占空间。换成Kryo序列化，通常能提速30%到50%。配置很简单：

spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") spark.conf.set("spark.kryo.registrationRequired", "false")

如果自定义了类，最好注册一下，否则Kryo会写入完整的类名，反而更慢。对于这个案例，用的都是基本类型和Row，不注册也没关系。

5.2 Shuffle分区数：200不是万能药

spark.sql.shuffle.partitions默认是200。对于小数据量，200个分区意味着每个分区只有几KB数据，任务启动开销比计算时间还长。对于大数据量，200个分区又太少，每个分区要处理几百MB，容易OOM。

我的经验公式：shuffle分区数 = shuffle数据量(MB) / 100，但不少于集群总核数。比如shuffle数据量是5GB，那就是5120MB / 100 = 51个分区，取整到60。如果集群有40个核，那就取60，保证每个核至少分到1到2个分区。

spark.conf.set("spark.sql.shuffle.partitions", "60")

注意这个参数是全局的，会影响所有shuffle操作。如果作业里有多个shuffle，且数据量差异很大，可以考虑用repartition或coalesce单独调整。

5.3 内存管理：Executor内存怎么分

Executor内存分三块：Execution Memory（用于shuffle、join、sort）、Storage Memory（用于缓存）、User Memory（用于自定义数据结构）。默认比例是0.5、0.5、0.4（实际是动态调整的）。如果shuffle数据量大，Execution Memory不够，就会溢写到磁盘，速度骤降。

我通常这样设置：

spark-submit \ --executor-memory 8g \ --executor-cores 4 \ --conf spark.memory.fraction=0.7 \ --conf spark.memory.storageFraction=0.3 \ ...

spark.memory.fraction=0.7表示70%的堆内存用于Execution和Storage，剩下30%给User Memory和预留。storageFraction=0.3表示Storage占这70%里的30%，也就是总堆的21%。如果作业里没有缓存RDD或DataFrame，可以把storageFraction调低到0.1，把更多内存留给Execution。

5.4 我的调优前后对比

同一份10GB数据，1000万用户，在3节点集群（每节点8核32G）上跑：

调优前的主要瓶颈是shuffle分区数太多，200个任务每个只处理50MB数据，但任务启动和调度开销占了大部分时间。调到60后，每个任务处理约170MB，计算时间占比提高，总耗时下降明显。另外Kryo序列化让shuffle数据量减少了约40%，进一步缩短了时间。

6. 常见问题与排查技巧实录

6.1 任务报错OutOfMemoryError怎么办

先看是Driver OOM还是Executor OOM。Driver OOM通常是collect()或take()把太多数据拉到本地了。比如你写了avgDf.collect()，而结果有1000万行，Driver内存直接爆。解决办法是用write把结果写到HDFS，或者用limit限制行数。

Executor OOM通常是单个分区的数据量太大。检查spark.sql.shuffle.partitions是否太小，或者是否有数据倾斜。临时解决办法是调大spark.executor.memory，但根本办法还是增加分区数或处理倾斜。

6.2 作业跑得特别慢，怎么定位瓶颈

第一步，打开Spark UI，看Stage页面。如果某个Stage的Shuffle Read特别大，说明shuffle是瓶颈。如果某个Task的Duration远大于其他Task，说明数据倾斜。如果所有Task都慢，但Shuffle Read不大，可能是CPU密集型的计算，比如复杂的UDF。

第二步，看explain()的输出。如果物理计划里有SortMergeJoin但数据量很小，可以改成BroadcastHashJoin。如果HashAggregate没有partial_前缀，说明部分聚合没生效，检查数据类型是否支持。

第三步，检查数据本地性。如果Locality Level显示ANY而不是NODE_LOCAL，说明数据没有在计算节点上，需要跨网络传输。这通常是因为HDFS block和Spark分区不对齐，可以通过repartition调整。

6.3 结果和单机算的不一样

最常见的原因是浮点数精度。Spark的avg()返回Double，单机Python的mean()也返回Double，但累加顺序不同会导致最后几位有差异。如果业务方要求完全一致，可以用BigDecimal做精确计算，但性能会下降。

另一个原因是空值处理。单机Pandas的mean()默认跳过NaN，Spark的avg()也跳过null，但如果数据里有空字符串，Pandas会报错，Spark会当成null。所以清洗规则要统一。

6.4 常见问题速查表

6.5 一个容易被忽略的坑：时间戳时区

心率数据的时间戳如果是UTC，而你按本地时间分组，会导致跨天数据错乱。比如北京时间早上8点的心率，UTC是凌晨0点，按UTC分组会算到前一天。解决办法是在读取时统一转成UTC，或者在分组时用from_utc_timestamp转换。这个坑我在实际项目中踩过，业务方反馈“早上的平均心率怎么跑到昨天去了”，查了半天才发现是时区问题。

7. 从平均心率到更多指标：扩展思路

平均心率只是最基础的统计量。实际业务中，医生或健康管理师更关心的是静息心率、最大心率、心率变异性（HRV）。这些指标用Spark也能算，只是逻辑复杂一些。

静息心率通常取睡眠期间的最低平均心率。可以先按用户和时间窗口（比如每5分钟）算平均，再取每天的最小值。最大心率就是max("heart_rate")，但要注意过滤掉异常高值。HRV需要计算相邻心跳间隔的标准差，这要求数据里有每次心跳的精确时间戳，而不是每分钟的平均值。

如果要做实时告警，可以把Spark Structured Streaming接上Kafka，每来一条心率记录就判断是否超过阈值。但Structured Streaming的延迟在秒级，对于心脏骤停这种紧急情况，还是得用Flink或者边缘计算在设备端直接判断。

我在实际项目中的体会是：批处理算历史指标，流处理做实时告警，两者配合使用。Spark在这个架构里承担的是“离线修正”和“模型训练”的角色，比如用历史数据训练一个异常心率检测模型，然后把模型部署到流处理任务里做在线推理。这样既发挥了Spark吞吐量大的优势，又避免了它在低延迟场景下的短板。

最后分享一个小技巧：如果数据量不大（比如几GB），但你又想用Spark的分布式能力，可以试试spark.sql.execution.arrow.pyspark.enabled=true，开启Arrow加速。Pandas UDF的传输效率比逐行序列化高一个数量级，尤其适合Python用户。我实测下来，同样的聚合逻辑，开了Arrow之后耗时从8分钟降到2分钟，效果非常明显。