DatabricksでのHyperloglog

# DatabricksでのHyperloglog

Hyperloglogは、webサイトのUU集計などでよく使われる確率的アルゴリズムを用いたデータのカーデナリティを推測するものになります。

細かくは、ブレインパッドさんの[ブログ]("https://blog.brainpad.co.jp/entry/2016/06/27/110000)やフライウィールさんの[ブログ]("https://www.flywheel.jp/topics/hyperloglog-sketch-in-practice/")に書いていただいていますのでそちらを参照されるのがいいかと思います。

# Hyperloglog関連の関数

Databricksでは、hyperloglog関連の関数として下記の関数が用意されています。

| Function | Description |
|:-:|:-:|
| approx_count_distinct(expr[,relativeSD])  |  グループ内の expr に含まれる明確な値の推定数を返します。|
|  approx_percentile(expr,percentage[,accuracy])| グループ内の expr のおおよそのパーセンタイルを返します。|
| approx_top_k(expr[,k[,maxItemsTracked]])  |  expr の中で最も頻繁に出現する上位 k 個の項目の値を、そのおおよその数とともに返します。 |

https://docs.databricks.com/sql/language-manual/sql-ref-functions-builtin.html#aggregate-functions

## approx_count_distinctの動作をみてみる
一番使いそうなapprox_count_distinctはこんな感じです。

テストデータセットから、9104511件ほどのuser idを集計してみます。

まずは、通常のdistinct_countです。
６.4秒ですね。

![Screenshot 2023-03-10 at 23.46.46.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/0f75a87e-9bd9-bc8e-ce23-4a13dd3aad0f.jpeg)

approx_count_distinctです。rsdはdefault 0.05です。
3.81秒です。件数は9312841なので、20万件ほどの差分がでます。

![Screenshot 2023-03-10 at 23.47.45.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/a505a71e-acfd-e886-0e62-e45a4be0fd51.jpeg)

rsdは0.01で設定します。
4.14秒です。件数は9007831なので、10万件ほどの差分がでます。

![Screenshot 2023-03-10 at 23.52.18.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/a35aa7f6-9129-f13c-3680-6b3d5d31e176.jpeg)

rsdは0.001で設定します。
4.95秒です。件数は9113218なので、8700件ほどの差分がでます。

![Screenshot 2023-03-10 at 23.52.18.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/f9c8688f-b44c-dced-33c8-955827ff4958.jpeg)

rdsを小さくしていくことで、精度はあがりますが、かかる時間が増えていくので注意が必要です。
また、spark公式Documentでは、0.01以下の場合はcountDistinctを使う方ほが効率的だと言っています。

https://spark.apache.org/docs/3.1.2/api/python/reference/api/pyspark.sql.functions.approx_count_distinct.html

## Sketchは・・・?
DatabricksのDefault関数では、あくまで推定値を返すのみのものになっており、sketchを保存して、さらにreaggregationをかけたい場合などには適しません。

そこで、活用できるのがswoop社が公開してくれているspark-alchemyです。

https://github.com/swoop-inc/spark-alchemy

https://www.slideshare.net/databricks/highperformance-advanced-analytics-with-sparkalchemy


SparkAI summit Europe2019でも登壇されていたように、hyperloglogでの集計をよりやりやすくするために多くのHLL関数を提供してくれます。(SparkやBigqueryよりも多くの関数を提供してくれています）

https://github.com/swoop-inc/spark-alchemy/wiki/Spark-HyperLogLog-Functions#spark-alchemys-approach-to-approximate-counting

## spark-alchemyの使い方

### Clusterへのinstall
MavenからClusterにinstallします。

![Screenshot 2023-03-11 at 0.05.13.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/2a5418fb-d5e7-2874-06a9-8c11f36d57ae.jpeg)

### library importします

```scala
import org.apache.spark.sql.functions._
import com.swoop.alchemy.spark.expressions.hll.functions._
```

### サンプルデータフレーム作成します
hll_init_aggでsketchのbinary生成します

```scala
spark.range(100000).createOrReplaceTempView("ids")

val df = (spark.range(100000).select(
  // exact distinct count
  countDistinct('id).as("cntd"),
  // Spark's HLL implementation with default 5% precision
  approx_count_distinct('id).as("anctd_spark_default"),
  // approximate distinct count with default 5% precision
  hll_cardinality(hll_init_agg('id)).as("acntd_default"),
  // generate sketch binary
  hll_init_agg('id).as("hll_binary"),
  // approximate distinct counts with custom precision
  map(
    Seq(0.005, 0.02, 0.05, 0.1).flatMap { error =>
      lit(error) :: hll_cardinality(hll_init_agg('id, error)) :: Nil
    }: _*
  ).as("acntd")
))
```

![Screenshot 2023-03-11 at 0.07.43.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/1661cd48-165e-7121-5d93-32625dc6baf4.jpeg)

### データフレームみてみる
こんな感じでsktchが保存されるので、再度集計するときにこのskecthをベース集計することで高速化が期待できます。

![Screenshot 2023-03-11 at 0.08.21.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/190656/d8de5c75-de15-a5d2-561d-23f80f25d276.jpeg)


また、こちらのドキュメントに再集計のtipsや高速化のためのサンプルまで書いていただいているのでぜひ参考にしてみてください。

https://github.com/swoop-inc/spark-alchemy/wiki/Spark-HyperLogLog-Functions#high-performance-distinct-counts-for-analytics