在 apache spark 中创建存储桶

Question

我想将数据存储在 spark 中，以使相差 5 秒或更短的时间戳与相应的数据一起落入一个 5 秒的存储桶中。同样，下一组 5 秒存储桶包含剩余的日志。 （这样我就可以汇总存储桶中的数据）。我的日志：

1472120400.107 HTTP GEO er.aujf.csdh.jkhydf.eyrgt
1472120399.999 HTTP GEO er.asdhff.cdn.qyirg.sdgsg
1472120397.633 HTTP GEO er.abff.kagsf.weyfh.ajfg
1472120397.261 HTTP GEO er.laffg.ayhrff.agyfr.yawr
1472120394.328 HTTP GEO er.qfryf.aqwruf.oiuqwr.agsf
1472120393.737 HTTP GEO er.aysf.aouf.ujaf.casf
.
.
.

我仍然无法弄清楚如何在火花中做到这一点。

时间戳为1472120400.107,1472120399.999,1472120397.633,1472120397.261等的日志落入一个bucket，下一组落入下一个bucket，以此类推。

输出：

时间戳为 1472120400.107,1472120399.999,1472120397.633,1472120397.261 的所有日志行都将保存在内存中（一个桶），以便进一步处理，例如查找整个桶的计数。同样，下一个桶。

Answer 1

只需将时间戳除以您要创建的粒度即可。在 PairRDD 中将 bin 编号作为 key，其中 data 是输入，然后是 reduceByKey。

我将在Scala中编写代码示例，基本上将其转换为python是微不足道的我想说明这一点。

val l5 = List("1472120400.107 HTTP GEO er.aujf.csdh.jkhydf.eyrgt", "1472120399.999 HTTP GEO er.asdhff.cdn.qyirg.sdgsg") 
val l5RDD = sc.parallelize(l5) //input as RDD
val l5tmp = l5RDD.map(item => item.split(" ")) //Split the sentence
val l5tmp2 = l5tmp.map(item => ((item(0).toDouble/3600000).toInt, List(item))) //Map the data to a bin (in the key) according to the wanted granularity
val collected = l5tmp2.reduceByKey(_ ++ _) //Collect the lists to create the bins of data
collected.collect().foreach(println) //Prints (408,List([Ljava.lang.String;@2c6aed22, [Ljava.lang.String;@e322ec9)) - means that both entries collected to a bin named 408