高效的计数与Apache Spark不同

Question

1亿个客户在几个网站的页面上点击1000亿次(比方说100个网站).点击流可在大型数据集中使用.

使用Apache Spark的抽象,每个网站计算不同访问者的最有效方法是什么？

Answer 1

visitors.distinct().count()这将是显而易见的方式,第一种方式是明确的,你可以指定并行水平,也可以看到速度的提高.如果可以将访问者设置为流并使用D流,则可以实时计算.您可以直接从目录流式传输,并使用与RDD相同的方法,如:

val file = ssc.textFileStream("...") file.distinct().count()

最后一个选项是使用,def countApproxDistinct(relativeSD: Double = 0.05): Long但是这被标记为实验性的,但如果relativeSD(标准偏差)更高,则会明显快于计数.

编辑:由于你想要每个网站的计数你可以减少网站ID,这可以有效地完成(使用组合器),因为计数是聚合的.如果您有一个RDD的网站名称用户ID元组,您可以这样做. visitors.countDistinctByKey()或者visitors.countApproxDistinctByKey(),大约一个是实验性的.要使用大约不同的键,您需要一个PairRDD

有趣的是,如果你对近似值没问题并想要快速的结果,你可能需要研究由与火花放大器实验室相同的人制作的blinkDB.

Answer 2

我必须做类似的事情,你可以做的一件有效的事情(那不是真正的火花)是将你的游客ID映射到字节列表而不是GUID字符串,你可以节省4倍的空间(因为2个字符是十六进制编码)一个字节,一个Char在一个字符串中使用2个字节).

// Inventing these custom types purely for this question - don't do this in real life!
type VistorID = List[Byte]
type WebsiteID = Int

val visitors: RDD[(WebsiteID, VisitorID)] = ???

visitors.distinct().mapValues(_ => 1).reduceByKey(_ + _)

请注意,您也可以这样做:

visitors.distinct().map(_._1).countByValue()

但这也没有扩展.

Answer 3

我注意到,在RDD上运行它时,基本的不同功能可以明显快于在DataFrame集合上运行它.例如:

DataFrame df = sqlContext.load(...)
df.distinct.count // 0.8 s
df.rdd.distinct.count // 0.2 s

Answer 4

如果data是RDD(站点,访问者)对,那么data.countApproxDistinctByKey(0.05)将给你一个RDD(站点,计数).可以减少参数以获得更多精度,但代价是更多处理.

Answer 5

Spark 2.0将ApproxCountDistinct添加到数据帧和SQL API中:

https://databricks.com/blog/2016/05/19/approximate-algorithms-in-apache-spark-hyperloglog-and-quantiles.html

https://spark.apache.org/docs/1.6.2/api/java/org/apache/spark/sql/functions.html#approxCountDistinct(org.apache.spark.sql.Column)