사용자 정의 집계 함수

사용자 정의 집계 함수 (User Defined Aggregate Functions, UDAFs)

UDAF(User-Defined Aggregate Function)는 여러 행에 한 번에 동작해 하나의 집계된 값을 결과로 반환하는 사용자 프로그래밍 루틴이에요. 이 문서는 UDAF를 생성하고 등록하는 데 필요한 클래스들과, Scala·Java에서 UDAF를 정의하고 등록한 뒤 Spark SQL에서 호출하는 예제를 담고 있어요.

출처: 문서

본문

설명 (Description)

사용자 정의 집계 함수(UDAFs)는 여러 행에 한 번에 동작해 하나의 집계된 값을 결과로 반환하는 사용자가 프로그래밍할 수 있는 루틴이에요. 이 문서는 UDAF를 생성하고 등록하는 데 필요한 클래스들을 나열해요. 또한 Scala에서 UDAF를 정의·등록하고 Spark SQL에서 호출하는 방법을 보여주는 예제도 포함하고 있어요.

Aggregator[-IN, BUF, OUT]

사용자 정의 집계를 위한 기반 클래스(base class)로, Dataset 연산에서 그룹의 모든 요소를 가져와 하나의 값으로 줄이는(reduce) 데 사용할 수 있어요.

IN - 집계의 입력 타입이에요.

BUF - 중간(reduction) 값의 타입이에요.

OUT - 최종 출력 결과의 타입이에요.

• bufferEncoder: Encoder[BUF]

  중간 값 타입에 대한 Encoder를 지정해요.

• finish(reduction: BUF): OUT

  reduction의 출력을 변환해요.

• merge(b1: BUF, b2: BUF): BUF

  두 개의 중간 값을 병합해요.

• outputEncoder: Encoder[OUT]

  최종 출력 값 타입에 대한 Encoder를 지정해요.

• reduce(b: BUF, a: IN): BUF

  입력 값 a를 현재 중간 값에 집계해요. 성능을 위해, 함수는 b에 대해 새 객체를 만들지 않고 b를 수정한 뒤 반환할 수 있어요.

• zero: BUF

  이 집계의 중간 결과 초기 값이에요.

예제 (Examples)

타입 안전한 사용자 정의 집계 함수 (Type-Safe User-Defined Aggregate Functions)

강한 타입(strongly typed)의 Dataset을 위한 사용자 정의 집계는 Aggregator 추상 클래스를 중심으로 이뤄져요. 예를 들어 타입 안전한 사용자 정의 평균은 다음과 같이 생겼어요:

import org.apache.spark.sql.{Encoder, Encoders, SparkSession}
import org.apache.spark.sql.expressions.Aggregator

case class Employee(name: String, salary: Long)
case class Average(var sum: Long, var count: Long)

object MyAverage extends Aggregator[Employee, Average, Double] {
  // A zero value for this aggregation. Should satisfy the property that any b + zero = b
  def zero: Average = Average(0L, 0L)
  // Combine two values to produce a new value. For performance, the function may modify `buffer`
  // and return it instead of constructing a new object
  def reduce(buffer: Average, employee: Employee): Average = {
    buffer.sum += employee.salary
    buffer.count += 1
    buffer
  }
  // Merge two intermediate values
  def merge(b1: Average, b2: Average): Average = {
    b1.sum += b2.sum
    b1.count += b2.count
    b1
  }
  // Transform the output of the reduction
  def finish(reduction: Average): Double = reduction.sum.toDouble / reduction.count
  // Specifies the Encoder for the intermediate value type
  def bufferEncoder: Encoder[Average] = Encoders.product
  // Specifies the Encoder for the final output value type
  def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

val ds = spark.read.json("examples/src/main/resources/employees.json").as[Employee]
ds.show()
// +-------+------+
// |   name|salary|
// +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

// Convert the function to a `TypedColumn` and give it a name
val averageSalary = MyAverage.toColumn.name("average_salary")
val result = ds.select(averageSalary)
result.show()
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0|
// +--------------+

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/sql/UserDefinedTypedAggregation.scala"에서 찾을 수 있어요.

import java.io.Serializable;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.TypedColumn;
import org.apache.spark.sql.expressions.Aggregator;

public static class Employee implements Serializable {
  private String name;
  private long salary;

  // Constructors, getters, setters...

}

public static class Average implements Serializable  {
  private long sum;
  private long count;

  // Constructors, getters, setters...

}

public static class MyAverage extends Aggregator<Employee, Average, Double> {
  // A zero value for this aggregation. Should satisfy the property that any b + zero = b
  @Override
  public Average zero() {
    return new Average(0L, 0L);
  }
  // Combine two values to produce a new value. For performance, the function may modify `buffer`
  // and return it instead of constructing a new object
  @Override
  public Average reduce(Average buffer, Employee employee) {
    long newSum = buffer.getSum() + employee.getSalary();
    long newCount = buffer.getCount() + 1;
    buffer.setSum(newSum);
    buffer.setCount(newCount);
    return buffer;
  }
  // Merge two intermediate values
  @Override
  public Average merge(Average b1, Average b2) {
    long mergedSum = b1.getSum() + b2.getSum();
    long mergedCount = b1.getCount() + b2.getCount();
    b1.setSum(mergedSum);
    b1.setCount(mergedCount);
    return b1;
  }
  // Transform the output of the reduction
  @Override
  public Double finish(Average reduction) {
    return ((double) reduction.getSum()) / reduction.getCount();
  }
  // Specifies the Encoder for the intermediate value type
  @Override
  public Encoder<Average> bufferEncoder() {
    return Encoders.bean(Average.class);
  }
  // Specifies the Encoder for the final output value type
  @Override
  public Encoder<Double> outputEncoder() {
    return Encoders.DOUBLE();
  }
}

Encoder<Employee> employeeEncoder = Encoders.bean(Employee.class);
String path = "examples/src/main/resources/employees.json";
Dataset<Employee> ds = spark.read().json(path).as(employeeEncoder);
ds.show();
// +-------+------+
// |   name|salary|
// +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

MyAverage myAverage = new MyAverage();
// Convert the function to a `TypedColumn` and give it a name
TypedColumn<Employee, Double> averageSalary = myAverage.toColumn().name("average_salary");
Dataset<Double> result = ds.select(averageSalary);
result.show();
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0|
// +--------------+

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/sql/JavaUserDefinedTypedAggregation.java"에서 찾을 수 있어요.

타입 없는 사용자 정의 집계 함수 (Untyped User-Defined Aggregate Functions)

위에서 설명한 타입 집계(typed aggregation)는 DataFrames와 함께 사용하기 위해 타입 없는(untyped) 집계 UDF로도 등록할 수 있어요. 예를 들어 타입 없는 DataFrames를 위한 사용자 정의 평균은 다음과 같이 생겼어요:

import org.apache.spark.sql.{Encoder, Encoders, SparkSession}
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.functions

case class Average(var sum: Long, var count: Long)

object MyAverage extends Aggregator[Long, Average, Double] {
  // A zero value for this aggregation. Should satisfy the property that any b + zero = b
  def zero: Average = Average(0L, 0L)
  // Combine two values to produce a new value. For performance, the function may modify `buffer`
  // and return it instead of constructing a new object
  def reduce(buffer: Average, data: Long): Average = {
    buffer.sum += data
    buffer.count += 1
    buffer
  }
  // Merge two intermediate values
  def merge(b1: Average, b2: Average): Average = {
    b1.sum += b2.sum
    b1.count += b2.count
    b1
  }
  // Transform the output of the reduction
  def finish(reduction: Average): Double = reduction.sum.toDouble / reduction.count
  // Specifies the Encoder for the intermediate value type
  def bufferEncoder: Encoder[Average] = Encoders.product
  // Specifies the Encoder for the final output value type
  def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

// Register the function to access it
spark.udf.register("myAverage", functions.udaf(MyAverage))

val df = spark.read.json("examples/src/main/resources/employees.json")
df.createOrReplaceTempView("employees")
df.show()
// +-------+------+
// |   name|salary|
// +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

val result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees")
result.show()
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0|
// +--------------+

전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/sql/UserDefinedUntypedAggregation.scala"에서 찾을 수 있어요.

import java.io.Serializable;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.expressions.Aggregator;
import org.apache.spark.sql.functions;

public static class Average implements Serializable  {
  private long sum;
  private long count;

  // Constructors, getters, setters...
  public Average() {
  }

  public Average(long sum, long count) {
    this.sum = sum;
    this.count = count;
  }

  public long getSum() {
    return sum;
  }

  public void setSum(long sum) {
    this.sum = sum;
  }

  public long getCount() {
    return count;
  }

  public void setCount(long count) {
    this.count = count;
  }
}

public static class MyAverage extends Aggregator<Long, Average, Double> {
  // A zero value for this aggregation. Should satisfy the property that any b + zero = b
  @Override
  public Average zero() {
    return new Average(0L, 0L);
  }
  // Combine two values to produce a new value. For performance, the function may modify `buffer`
  // and return it instead of constructing a new object
  @Override
  public Average reduce(Average buffer, Long data) {
    long newSum = buffer.getSum() + data;
    long newCount = buffer.getCount() + 1;
    buffer.setSum(newSum);
    buffer.setCount(newCount);
    return buffer;
  }
  // Merge two intermediate values
  @Override
  public Average merge(Average b1, Average b2) {
    long mergedSum = b1.getSum() + b2.getSum();
    long mergedCount = b1.getCount() + b2.getCount();
    b1.setSum(mergedSum);
    b1.setCount(mergedCount);
    return b1;
  }
  // Transform the output of the reduction
  @Override
  public Double finish(Average reduction) {
    return ((double) reduction.getSum()) / reduction.getCount();
  }
  // Specifies the Encoder for the intermediate value type
  @Override
  public Encoder<Average> bufferEncoder() {
    return Encoders.bean(Average.class);
  }
  // Specifies the Encoder for the final output value type
  @Override
  public Encoder<Double> outputEncoder() {
    return Encoders.DOUBLE();
  }
}

// Register the function to access it
spark.udf().register("myAverage", functions.udaf(new MyAverage(), Encoders.LONG()));

Dataset<Row> df = spark.read().json("examples/src/main/resources/employees.json");
df.createOrReplaceTempView("employees");
df.show();
// +-------+------+
// |   name|salary|
// +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

Dataset<Row> result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees");
result.show();
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0|
// +--------------+

전체 예제 코드는 Spark 저장소의 "examples/src/main/java/org/apache/spark/examples/sql/JavaUserDefinedUntypedAggregation.java"에서 찾을 수 있어요.

-- Compile and place UDAF MyAverage in a JAR file called `MyAverage.jar` in /tmp.
CREATE FUNCTION myAverage AS 'MyAverage' USING JAR '/tmp/MyAverage.jar';

SHOW USER FUNCTIONS;
+------------------+
|          function|
+------------------+
| default.myAverage|
+------------------+

CREATE TEMPORARY VIEW employees
USING org.apache.spark.sql.json
OPTIONS (
    path "examples/src/main/resources/employees.json"
);

SELECT * FROM employees;
+-------+------+
|   name|salary|
+-------+------+
|Michael|  3000|
|   Andy|  4500|
| Justin|  3500|
|  Berta|  4000|
+-------+------+

SELECT myAverage(salary) as average_salary FROM employees;
+--------------+
|average_salary|
+--------------+
|        3750.0|
+--------------+

관련 구문 (Related Statements)

• Scalar User Defined Functions (UDFs)
• Integration with Hive UDFs/UDAFs/UDTFs

더 알아보기 (Learn more)

• 아파치 스파크 SQL 참조 (원문)
• Scalar User-Defined Functions (UDFs) — 스칼라 UDF
• Integration with Hive UDFs/UDAFs/UDTFs — Hive UDF 통합
• Functions — 함수 전반 문서