如何用Spark（Scala）处理增量销售数据并生成周度平均销售额报表

Hey there! Let's break down how to solve this Spark + Scala weekly batch problem, especially handling those tricky data updates that keep popping up. The key here is building idempotent pipelines—meaning re-running the job won't mess up your results even if data gets updated after your first run. Here's a step-by-step solution:

1. First: Secure Your Daily Data with Version Control #

The biggest pain point with data updates is ensuring you always have the latest version of each day's sales data. Instead of overwriting files (which is risky), use Delta Lake (built for Spark) to store your daily sales data with ACID compliance and version history. This lets you safely update old records without losing track of changes.

Example Code for Ingesting & Updating Daily Feeds #

import org.apache.spark.sql.functions._
import io.delta.tables._

// Read the daily sales feed (adjust format/schema to match your actual data)
val dailySalesRaw = spark.read.format("csv")
  .option("header", "true")
  .schema("Day date, product string, sales double")
  .load("/path/to/daily-sales-feed")
  .withColumn("last_updated", current_timestamp()) // Track when this record was added/updated

// Connect to your Delta table (creates it if it doesn't exist)
val dailySalesDelta = DeltaTable.forPath(spark, "/path/to/daily-sales-delta-store")

// Merge logic: If we already have a record for (Day, product), update it with the latest sales value
// If not, insert the new record
dailySalesDelta.as("target")
  .merge(
    dailySalesRaw.as("source"),
    "target.Day = source.Day AND target.product = source.product"
  )
  .whenMatchedUpdateExpr(Map(
    "sales" -> "source.sales",
    "last_updated" -> "source.last_updated"
  ))
  .whenNotMatchedInsertAll()
  .execute()

2. Calculate Weekly Average Sales (Using Trusted Daily Data) #

Now that your daily data is always up-to-date, compute the weekly average. First, make sure you're aggregating daily totals per product (in case a day has multiple entries for the same product), then calculate the average across the week.

Example Code for Weekly Aggregation #

// First, get the latest daily totals per product (in case of multiple entries per day)
val dailyAggregated = dailySalesDelta.toDF()
  .groupBy("Day", "product")
  .agg(sum("sales").alias("daily_total"))

// Calculate weekly average: Use date_trunc to get the start of the week as your "Week" identifier
val weeklyAvg = dailyAggregated
  .withColumn("Week", date_trunc("week", col("Day")).cast("date")) // Format week as a date (e.g., 2024-05-20 for the week starting Monday)
  .groupBy("Week", "product")
  .agg(avg("daily_total").alias("sales_average"))
  .orderBy("Week", "product")

3. Merge with Historical Weekly Results #

Just like with daily data, you need to update your historical weekly results if a day's data changes (which would alter the week's average). Again, Delta Lake makes this easy with merge logic.

Example Code for Merging Historical Data #

// Connect to your existing weekly results Delta table
val weeklyResultsDelta = DeltaTable.forPath(spark, "/path/to/weekly-results-delta-store")

// Merge the new weekly averages into the historical table:
// Update existing (Week, product) entries with the new average, insert new ones
weeklyResultsDelta.as("target")
  .merge(
    weeklyAvg.as("source"),
    "target.Week = source.Week AND target.product = source.product"
  )
  .whenMatchedUpdateExpr(Map(
    "sales_average" -> "source.sales_average"
  ))
  .whenNotMatchedInsertAll()
  .execute()

// Export the final merged results to your desired file format (CSV/Parquet)
weeklyResultsDelta.toDF()
  .write.format("csv")
  .option("header", "true")
  .mode("overwrite") // Safe because Delta already maintains the latest state
  .save("/path/to/final-weekly-output")

4. Bonus Tips to Make This Robust #

• Partition Your Delta Tables: Partition the daily table by Day and the weekly table by Week—this will speed up queries by only scanning relevant data.
• Add Checks: Before running the weekly aggregation, check if any data in the target week has been updated (using Delta's version history). If no changes, skip the aggregation to save resources.
• Log Everything: Track when updates happen, which days were modified, and how weekly averages changed. This helps debug if something goes wrong.

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