spark-alchemy中的HyperLogLog功能使用实践

简介

利用HyperLogLog算法计算UV类指标是一种常用的方案。在spark中的approx_count_distinct函数就是基于HyperLogLog实现的，但其每次都需要从原始明细的数据进行计算，无法从中间结果进行再聚合。预先聚合是一种常用高性能分析的手段，可以极大地减少数据量。由于Spark没有提供相应功能，Swoop 开源了高性能的HLL native函数工具包，作为 spark-alchemy 项目的一部分，具体使用示例可以参考 HLL docs。spark-alchemy中的HLL sketch结构是可再聚合的，可以将其存储下来，通过预聚合大幅减少了要处理的数据量。

添加maven包依赖

在项目pom中添加maven包依赖：

<!-- https://mvnrepository.com/artifact/com.swoop/spark-alchemy -->
<dependency>
    <groupId>com.swoop</groupId>
    <artifactId>spark-alchemy_2.11</artifactId>
    <version>0.5.5</version>
</dependency>

准备测试数据

参考HLL docs中的例子准备测试数据：

spark.range(100000).toDF("id").createOrReplaceTempView("ids")

创建Hive表存储中间结果

从HLL函数的文档中可知，使用hll_init_agg函数可创建中间结果HLL sketch。其是一个二进制数组结构，因此可建如下测试表：

CREATE TABLE t_hllp_alchemy_test (
  id_mod bigint, 
  hll_id binary
)

SparkSQL向表中写入数据

可通过SparkSQL的方式写入数据：

// Register spark-alchemy HLL functions for use from SparkSQL
com.swoop.alchemy.spark.expressions.hll.HLLFunctionRegistration.registerFunctions(spark)
spark.sql("insert overwrite table t_hllp_alchemy_test select id % 10 as id_mod, hll_init_agg(id) as hll_id from ids group by id % 10")

从表中读数据进行计算

读表数据并计算其基数值：

spark.sql("select id_mod, hll_cardinality(hll_id) as acntd from t_hllp_alchemy_test order by id_mod").show()

+------+-----+
|id_mod|acntd|
+------+-----+
|     0| 9554|
|     1| 9884|
|     2| 9989|
|     3|10159|
|     4| 9987|
|     5| 9770|
|     6| 9921|
|     7| 9774|
|     8| 9414|
|     9| 9390|
+------+-----+

读表数据并进行再聚合计算：

spark.sql("select id_mod % 2 as id_mod2,  hll_cardinality(hll_merge(hll_id)) as acntd from t_hllp_alchemy_test group by id_mod % 2")

+-------+-----+
|id_mod2|acntd|
+-------+-----+
|      0|47305|
|      1|53156|
+-------+-----+

如不想使用二进制的列，也可将二进制数据转为十六进制的字符串进行存储：

spark.sql("insert overwrite table t_hllp_alchemy_test select id % 10 as id_mod10, hex(hll_init_agg(id)) as hll_id from ids group by id % 10")

然后从Hive表中读出数据并恢复成二进制数组数据进行计算：

spark.sql("select id_mod, hll_cardinality(unhex(hll_id)) as acntd from t_hllp_alchemy_test order by id_mod").show()

+------+-----+
|id_mod|acntd|
+------+-----+
|     0| 9554|
|     1| 9884|
|     2| 9989|
|     3|10159|
|     4| 9987|
|     5| 9770|
|     6| 9921|
|     7| 9774|
|     8| 9414|
|     9| 9390|
+------+-----+