UDF(사용자 정의 집계 함수)

적용 대상: Databricks Runtime으로 표시됨

UDF(사용자 정의 집계 함수)는 한 번에 여러 행에서 작동하고 결과적으로 단일 집계된 값을 반환하는 사용자 프로그래밍 가능한 루틴입니다. 이 설명서에는 UDAF를 만들고 등록하는 데 필요한 클래스가 나열되어 있습니다. 또한 Scala에서 UDAF를 정의 및 등록하고 Spark SQL에서 호출하는 방법을 보여 주는 예제도 포함되어 있습니다.

집계

구문Aggregator[-IN, BUF, OUT]

데이터 세트 작업에서 그룹의 모든 요소를 가져와 단일 값으로 줄이는 데 사용할 수 있는 사용자 정의 집계에 대한 기본 클래스입니다.

• IN: 집계에 대한 입력 형식입니다.

• BUF: 감소의 중간 값 형식입니다.

• OUT: 최종 출력 결과의 형식입니다.

• bufferEncoder: Encoder[BUF]

  중간 값 형식의 인코더입니다.

• finish(reduction: BUF): OUT

  감소의 출력을 변환합니다.

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

  두 중간 값을 병합합니다.

• outputEncoder: Encoder[OUT]

  최종 출력 값 형식의 인코더입니다.

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

  입력 값을 현재 중간 값 a 으로 집계합니다. 성능을 위해 함수는 에 대한 b새 개체를 생성하는 대신 수정 b 하고 반환할 수 있습니다.

• 0: BUF

  이 집계에 대한 중간 결과의 초기 값입니다.

예

형식 안전 사용자 정의 집계 함수

강력한 형식의 데이터 세트에 대한 사용자 정의 집계는 추상 클래스를 Aggregator 중심으로 진행됩니다. 예를 들어 형식이 안전한 사용자 정의 평균은 다음과 같습니다.

형식화되지 않은 사용자 정의 집계 함수

위에서 설명한 대로 형식화된 집계는 DataFrames에 사용하기 위해 형식화되지 않은 집계 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
  // The Encoder for the intermediate value type
  def bufferEncoder: Encoder[Average] = Encoders.product
  // 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.format("json").load("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|
// +--------------+

Java

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 static class MyAverage extends Aggregator<Long, Average, Double> {
  // A zero value for this aggregation. Should satisfy the property that any b + zero = b
  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
  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
  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
  public Double finish(Average reduction) {
    return ((double) reduction.getSum()) / reduction.getCount();
  }
  // The Encoder for the intermediate value type
  public Encoder<Average> bufferEncoder() {
    return Encoders.bean(Average.class);
  }
  // The Encoder for the final output value type
  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().format("json").load("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|
// +--------------+

Sql

-- 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|
+--------------+

관련 문

• UDF(스칼라 사용자 정의 함수)
• Hive UDF, UDAF 및 UDF와 통합