如何使用Spark/Scala持久化每日批处理数据以实现周度聚合？

嘿，这个场景我太熟了！之前用Spark+Scala做过类似的零售销售批处理系统，刚好能给你一套落地的方案。核心就是把每日转换好的数据持久化下来，周度任务直接复用，完全不用重复跑一遍原始数据的导入和转换，省超多资源！

每日任务负责：原始销售文件导入 → 数据转换 → 持久化转换结果；周度任务直接读取这些持久化的数据，执行聚合计算，完美规避重复劳动。

1. 每日批处理：导入、转换与持久化 #

首先，因为每日文件Schema一致，我们可以提前定义好固定Schema，避免Spark自动推断带来的开销和错误。然后选择Parquet作为持久化格式——它是Spark的最佳拍档：列存储压缩率高、读写速度快，还能完整保留Schema，后续读取根本不用重新定义结构。

下面是Scala代码示例，我给你加了详细注释：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object DailySalesProcessor {
  def main(args: Array[String]): Unit = {
    // 初始化SparkSession
    val spark = SparkSession.builder()
      .appName("DailySalesProcessing")
      .getOrCreate()
    
    // 提前定义固定Schema（和销售文件结构匹配）
    val salesSchema = "transaction_id STRING, sale_date DATE, product_id STRING, quantity INT, amount DOUBLE"
    
    // 读取当日销售文件：建议从命令行传入文件路径，方便调度工具传参
    val inputPath = args(0) // 比如 "/data/sales/raw/sales_20240520.csv"
    val rawSalesDF = spark.read
      .schema(salesSchema)
      .option("header", "true")
      .csv(inputPath)
    
    // 执行你的转换操作：这里举几个常见例子，你可以替换成自己的逻辑
    val transformedSalesDF = rawSalesDF
      // 过滤无效数据（比如数量或金额为负）
      .filter(col("quantity") > 0 && col("amount") > 0)
      // 计算单交易总金额
      .withColumn("transaction_total", col("quantity") * col("amount"))
      // 添加处理日期标记，方便后续追溯
      .withColumn("processing_date", current_date())
    
    // 持久化转换后的数据：按销售日期分区存储，周度聚合时能快速过滤
    val outputBasePath = "/data/sales/persisted/daily_transformed"
    transformedSalesDF.write
      .mode("append") // 每日追加新的分区
      .partitionBy("sale_date") // 按日期分区是核心优化点
      .parquet(outputBasePath)
    
    spark.stop()
  }
}

2. 周度聚合：读取持久化数据计算 #

周度任务就简单多了——直接读取Parquet文件，不用碰原始数据，也不用重复执行转换逻辑。如果数据量很大，还可以通过日期范围过滤只读取本周的分区，大幅减少数据扫描量。

代码示例：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object WeeklySalesAggregator {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("WeeklySalesAggregation")
      .getOrCreate()
    
    // 读取所有每日持久化的转换数据
    val persistedPath = "/data/sales/persisted/daily_transformed"
    val dailyTransformedDF = spark.read.parquet(persistedPath)
    
    // 过滤本周的数据：同样建议从命令行传入起止日期
    val weekStartDate = args(0) // 比如 "2024-05-13"
    val weekEndDate = args(1) // 比如 "2024-05-19"
    val weeklySalesDF = dailyTransformedDF
      .filter(col("sale_date").between(weekStartDate, weekEndDate))
      // 按产品ID做周度聚合，你可以根据需求调整维度
      .groupBy("product_id")
      .agg(
        sum("quantity").alias("weekly_total_qty"),
        sum("transaction_total").alias("weekly_total_amount"),
        countDistinct("transaction_id").alias("weekly_transaction_count")
      )
    
    // 输出周度结果：可以写入CSV、数据库或者数据仓库
    weeklySalesDF.write
      .mode("overwrite")
      .option("header", "true")
      .csv("/data/sales/reports/weekly_sales_20240519.csv")
    
    spark.stop()
  }
}

3. 几个实用优化建议 #

• 数据质量校验：每日任务里加一步校验，比如检查transaction_id是否唯一、sale_date是否为当日，避免坏数据污染持久化存储
• 分区优化：如果数据量特别大，除了按sale_date分区，还可以加个二级分区（比如product_category），进一步提升周度聚合的查询速度
• 任务调度：用Airflow、Oozie或者Spark自带的调度工具，自动触发每日任务（比如凌晨2点）和每周任务（比如周日晚上），完全不用手动操作
• 临时缓存：如果每日转换后需要多次使用数据，可以临时用transformedSalesDF.persist()缓存到内存+磁盘，但长期持久化还是要落地到Parquet，因为集群重启后缓存会丢失

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