今天先是把手动管理的 offset 从本地文件换成了简单的 Redis 存储，比存本地更贴合实际，代码里加了 Redis 的读写方法，消费完 Kafka 数据后把每个分区的 offset 存到 Redis，程序重启时先从 Redis 读 offset 再开始消费，避免了本地文件在分布式环境下的问题，过程中踩了 Redis 连接超时的坑，改了配置里的超时时间就好了。然后在流处理的流程里加了数据清洗步骤，对接收到的 Kafka 消息做了非空过滤、特殊字符剔除，还把时间字段做了格式转换，用 DStream 的 map 和 filter 算子就能实现，和批处理的清洗逻辑差不多，加了清洗后发现统计结果里的无效数据少了很多。另外还试了把流处理的最终统计结果输出到 Parquet 文件和控制台，用 saveAsTextFiles 指定输出路径，设置了按时间切分文件，方便后续查看，就是一开始没设置输出模式，导致重复输出报错，加了覆盖模式就解决了。
