在 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页左右大数据面试宝典 ，回复“大数据”获取多本大数据电子书