spark sql多个指标合并为1条sql

在 Spark SQL 指标开发中，使用一条 SQL 语句来计算多个指标是常见的做法。以下是几种常用的方法：

使用子查询：将多个指标的计算逻辑放在子查询中，再在外层 SELECT 语句中进行聚合。这种方式可以避免代码冗余，但可能会降低 SQL 语句的可读性。
使用 Window 函数：使用 Window 函数可以在一条 SQL 语句中计算多个指标，并且可以避免对同一个数据集进行多次扫描。具体操作是使用 over() 函数配合 sum、avg、max、min等聚合函数。这种方式相比子查询可以提升 SQL 的效率，但需要熟练掌握 Window 函数的使用方法。
使用 UDAF：如果需要计算的指标比较复杂，可以考虑开发自定义聚合函数（UDAF），这样可以将计算逻辑封装到统一的函数中，提高代码的可复用性和可维护性。

例子：

使用子查询进行多指标计算

假设我们有一个订单表 orders，包含以下字段：order_id, customer_id, order_date, amount。我们需要计算每个客户在 2022 年的订单数量和总订单金额。可以使用如下 SQL 语句实现：

SELECT 
  customer_id, 
  (SELECT COUNT(*) 
   FROM orders 
   WHERE customer_id = o.customer_id AND YEAR(order_date) = 2022) AS order_count, 
  (SELECT SUM(amount) 
   FROM orders 
   WHERE customer_id = o.customer_id AND YEAR(order_date) = 2022) AS total_amount 
FROM orders o 
GROUP BY customer_id;

这个 SQL 语句使用了两个子查询来计算每个客户在 2022 年的订单数量和总订单金额。子查询返回的结果会作为 select 语句中的一个列，因此可以使用 group by 对客户进行分组。

使用 Window 函数进行多指标计算

使用 Window 函数可以更方便地对查询结果进行分析。假设我们有一个销售表 sales，包含以下字段：sale_id, customer_id, product_id, sale_date, sale_amount。我们需要计算以下指标：每个客户的总销售额、每个客户的最大销售额、每个客户的销售额排名。可以使用如下 SQL 语句实现：

SELECT 
  customer_id, 
  SUM(sale_amount) OVER (PARTITION BY customer_id) AS total_sale, 
  MAX(sale_amount) OVER (PARTITION BY customer_id) AS max_sale, 
  RANK() OVER (PARTITION BY customer_id ORDER BY sale_amount DESC) AS sale_rank 
FROM sales;

这个 SQL 语句使用了三个 Window 函数，分别计算每个客户的总销售额、最大销售额和销售额排名。这里我们将结果按客户分组，然后使用了 Partition By 子句指定了客户维度。

使用自定义聚合函数进行多指标计算

自定义聚合函数（UDAF）可以针对特定的需求编写自己的聚合逻辑。以计算最小值、最大值和平均值为例，我们可以实现一个自定义平均值 UDAF：

import org.apache.spark.sql.expressions.MutableAggregationBuffer
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
 
class CustomAvg extends UserDefinedAggregateFunction {
 
  // 输入参数的数据类型
  def inputSchema: StructType = new StructType().add("inputColumn", DoubleType)
 
  // 中间结果的数据类型
  def bufferSchema: StructType = new StructType().add("sum", DoubleType).add("count", LongType)
 
  // 输出结果的数据类型
  def dataType: DataType = DoubleType
 
  // 是否固定输入参数类型
  def deterministic: Boolean = true
 
  // 初始化中间结果缓存
  def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0.0 // sum
    buffer(1) = 0L // count
  }
 
  // 更新中间结果缓存
  def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    if (!input.isNullAt(0)) {
      buffer(0) = buffer.getDouble(0) + input.getDouble(0)
      buffer(1) = buffer.getLong(1) + 1
    }
  }
 
  // 合并中间结果缓存
  def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getDouble(0) + buffer2.getDouble(0)
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }
 
  // 计算最终结果
  def evaluate(buffer: Row): Any = {
    buffer.getDouble(0) / buffer.getLong(1)
  }
}

这里我们实现了一个 CustomAvg UDAF，可以通过输入若干个 Double 类型的值，计算它们的平均值并返回。我们可以在 SparkSQL 中使用这个 UDAF 计算多个指标的平均值：

SELECT 
  CustomAvg(column1) AS avg1, 
  CustomAvg(column2) AS avg2, 
  CustomAvg(column3) AS avg3 
FROM table;

这个 SQL 语句使用了 CustomAvg UDAF 计算了三个列的平均值，可以根据实际需求进行扩展。

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书