[Spark] approxCountDistinct 알아보기

 

- 목차

 

들어가며.

이번 글에서는 SparkSQL 의 approxCountDistinct 함수에 대해 알아보려고 합니다.

approxCountDistinct 은 특정 칼럼의 값들이 몇개의 범주로 구성되어있는지를 파악하는 함수입니다.

즉, Categorical Data 의 종류 갯수를 파악합니다.

approxCountDistinct 는 이름에서 유추할 수 있듯이, 대략적인 결과를 획득하기 위한 함수인데요.

큰 규모의 데이터셋에 대해서 약간의 오차를 허용하고 결과 획득 시간을 단축하기 위한 목적을 가집니다.

 

사용할 데이터.

사용할 데이터는 Kaggle 의 Bank Marketing 데이터를 사용합니다.

아래 링크에서 다운로드받으실 수 있습니다.

https://www.kaggle.com/datasets/henriqueyamahata/bank-marketing

Bank Marketing

 

approxCountDistinct.

SparkSQL 에서는 Count 와 관련된 여러가지 함수들이 존재합니다.

count, countDistinct, approxCountDistinct 등이 존재하는데요.

count 와 countDistinct 는 DataFrame 의 정확한 사이즈를 확인할 수 있는 반면

approx_count_distinct 함수는 SparkSQL 에서 제공되는 DataFrame 의 대략적인 사이즈를 유추할 수 있습니다.

 

활용 예시는 아래와 같습니다.

from pyspark.sql import SparkSession
from pyspark.sql.functions import approxCountDistinct, countDistinct, count, col
from pyspark import SparkConf

conf = SparkConf()
conf.setMaster("local[*]")
conf.setAppName("test_count_action")
conf.set("spark.driver.bindAddress", "localhost")

spark = SparkSession.builder.config(conf=conf).getOrCreate()
df = (spark.read.format("csv")
      .option("delimiter", ";")
      .option("header", True)
      .option("inferSchema", True)
      .load("./dataset/bank-additional-full.csv"))

count_rows = df.select(count(col("*"))).collect()
distinctCount_rows = df.select(countDistinct(col("age"))).collect()
approxCountDistinct_rows = df.select(approxCountDistinct(col("age"), 0.1)).collect()
print(f"count : {count_rows[0]} \n")
print(f"distinctCount : {distinctCount_rows[0]} \n")
print(f"approxCountDistinct : {approxCountDistinct_rows[0]} \n")
spark.stop()

count : Row(count(1)=41188) 

distinctCount : Row(count(DISTINCT age)=78) 

approxCountDistinct : Row(approx_count_distinct(age)=85)

 

위 결과처럼 DataFrame 은 총 41188 개의 Row 들을 가집니다.

그리고 "age" 칼럼의 Distinct Count 은 78 개이고,

approxCountDistinct 값은 85 을 가집니다.

이는 어느 정도의 측정 오차를 감안하고 빠른 속도로 결과를 획득할 수 있는 이점을 제공합니다.

큰 규모의 데이터셋에서 그 효과를 발휘합니다.

 

아래는 검산은 위해서 age 의 고유값들을 조회하는 내용입니다.

총 78개의 값들이 존재합니다.

distinct_age_rows = df.select(col("age")).distinct().collect()
distinct_ages = sorted(list(map(lambda x: x[0], distinct_age_rows)))
print(f"distinct_ages : {distinct_ages}")

distinct_ages : [17, 18, 19, 20, 21, 22, 
23, 24, 25, 26, 27, 28, 29, 30, 31, 
32, 33, 34, 35, 36, 37, 38, 39, 40, 
41, 42, 43, 44, 45, 46, 47, 48, 49, 
50, 51, 52, 53, 54, 55, 56, 57, 58, 
59, 60, 61, 62, 63, 64, 65, 66, 67, 
68, 69, 70, 71, 72, 73, 74, 75, 76, 
77, 78, 79, 80, 81, 82, 83, 84, 85, 
86, 87, 88, 89, 91, 92, 94, 95, 98]

 

RSD (Relative Standard Deviation).

approxCountDistinct 함수는 rsd 라는 오차 정도를 지정할 수 있습니다.

rsd 의 값이 클수록 결과를 획득하는 속도가 빨라지고, 오차 정도가 증가합니다.

 

4GB 사이즈 && Row Count : 125,000,475 의 데이터를 기준으로 수행한 벤치마크 결과를 아래와 같습니다.

countDistinct 보다 approxCountDistinct 의 속도가 더 빠르긴하지만,

RSD 가 매우 낮으면 퍼포먼스가 하향되는 경향을 보입니다.

countDistinct.

elapsed : 0:00:13.641906

 

approxCountDistinct with RSD : 0.01

 elapsed : 0:00:14.525410

 

approxCountDistinct with RSD : 0.3

elapsed : 0:00:11.607573