Spark/Scala中不使用functions._实现客户账户数据聚合计算

需求说明 #

在Spark/Scala环境下，不引入org.apache.spark.sql.functions._，仅使用Dataset API完成针对客户与账户数据的聚合操作，具体包括：

• 按客户汇总账户余额总和（totalBalance）
• 统计客户名下的账户数量（numberAccounts）
• 计算账户余额平均值（averageBalance）
• 收集客户名下所有账户的AccountData序列（结构为{customerId, accountId, balance}）

数据字段定义：

• 客户数据：customerId、forename、surname
• 账户数据：customerId、accountId、balance

纯Dataset API实现代码 #

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Dataset, SparkSession}

object Assignment1 extends App {
  // 创建Spark会话
  val spark = SparkSession.builder().master("local[*]").appName("Assignment").getOrCreate()

  import spark.implicits._
  // 设置日志级别为OFF
  Logger.getRootLogger.setLevel(Level.OFF)

  // 定义数据结构
  case class CustomerData(customerId: String, forename: String, surname: String)
  case class AccountData(customerId: String, accountId: String, balance: Long)
  case class CustomerAccountOutput(
    customerId: String,
    forename: String,
    surname: String,
    accounts: Seq[AccountData],
    numberAccounts: Int,
    totalBalance: Long,
    averageBalance: Double
  )

  // 读取并转换客户数据源
  val customerDS: Dataset[CustomerData] = spark.read
    .option("header", "true")
    .option("sep", "\t")
    .csv("./dataset/customer_data.csv")
    .as[CustomerData]

  // 读取并转换账户数据源
  val accountDS: Dataset[AccountData] = spark.read
    .option("header", "true")
    .option("sep", "\t")
    .csv("./dataset/account_data.csv")
    .map(row => AccountData(
      row.getAs[String]("customerId"),
      row.getAs[String]("accountId"),
      row.getAs[String]("balance").toLong
    ))

  // 执行聚合操作
  val customerAccountDS: Dataset[CustomerAccountOutput] = customerDS
    // 左连接客户与账户数据，返回(CustomerData, Option[AccountData])类型的Dataset
    .joinWith(accountDS, customerDS.customerId === accountDS.customerId, "left_outer")
    // 按客户对象分组
    .groupByKey(_._1)
    // 遍历每组计算聚合指标
    .mapGroups { case (customer, accountsIter) =>
      // 提取非空的账户数据列表
      val accountsList = accountsIter.flatMap(_._2).toSeq
      // 计算各项聚合值
      val numberAccounts = accountsList.size
      val totalBalance = accountsList.map(_.balance).sum
      val averageBalance = if (numberAccounts > 0) totalBalance.toDouble / numberAccounts else 0.0
      // 构造输出结果对象
      CustomerAccountOutput(
        customer.customerId,
        customer.forename,
        customer.surname,
        accountsList,
        numberAccounts,
        totalBalance,
        averageBalance
      )
    }

  // 输出结果结构与数据
  customerAccountDS.printSchema()
  customerAccountDS.show(truncate = false)
}

关键实现说明 #

1. 数据转换：账户数据读取时直接通过map将字符串类型的balance转为Long，避免依赖Spark SQL的类型转换函数。
2. 左连接处理：使用joinWith替代DataFrame的join，直接得到带类型的连接结果，保留Dataset的类型安全特性。
3. 分组聚合：通过groupByKey按客户分组，再用mapGroups手动遍历每组数据计算聚合指标：
   • 用flatMap(_._2)过滤左连接产生的空账户记录
   • 基于账户列表直接统计数量、求和、计算平均值
4. 空值兼容：当客户无账户时，平均值直接设为0，无需额外调用空值填充函数。

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