我必须使用apache spark和scala作为编程语言在数据集上执行以下任务：
从hdfs读取数据集。一些采样线如下所示：
deviceid,bytes,eventdate
15590657,246620,20150630
14066921,1907,20150621
14066921,1906,20150626
6522013,2349,20150626
6522013,2525,20150613按设备id对数据进行分组。因此，我们现在有了deviceid=>（字节，eventdate）的Map
对于每个设备，按事件日期对集合进行排序。我们现在为每个设备提供了一组基于事件日期的有序字节。
从这个有序集合中选取最后30天的字节。
使用30的时间段查找最后一个日期的移动平均字节数。
使用30的时间段查找最终日期字节的标准偏差。
返回结果中的两个值（mean-kstddev）和（mean+kstddev）[假设k=3]
我使用的是ApacheSpark1.3.0。实际的数据集更宽，最终必须在10亿行上运行。
以下是数据集的数据结构。
package com.testing
case class DeviceAggregates (
device_id: Integer,
bytes: Long,
eventdate: Integer
) extends Ordered[DailyDeviceAggregates] {
def compare(that: DailyDeviceAggregates): Int = {
eventdate - that.eventdate
}
}
object DeviceAggregates {
def parseLogLine(logline: String): DailyDeviceAggregates = {
val c = logline.split(",")
DailyDeviceAggregates(c(0).toInt, c(1).toLong, c(2).toInt)
}
}deviceanalyzer类如下所示：
package com.testing
import com.testing.DeviceAggregates
import org.apache.spark.{SparkContext, SparkConf}
import scala.util.Sorting
object DeviceAnalyzer {
def main(args: Array[String]) {
val sparkConf = new SparkConf().setAppName("Device Statistics Analyzer")
val sc = new SparkContext(sparkConf)
val logFile = args(0)
val deviceAggregateLogs = sc.textFile(logFile).map(DeviceAggregates.parseLogLine).cache()
val deviceIdsMap = deviceAggregateLogs.groupBy(_.device_id)
deviceIdsMap.foreach(
// I am stuck here !!
})
sc.stop()
}
}但我仍然坚持这个算法的实际实现。
1条答案
按热度按时间6pp0gazn1#
我有一个非常粗糙的实现来完成这项工作，但它不符合标准。对不起，我对scala/spark非常陌生，所以我的问题很基本。以下是我现在拥有的：
如果有人能提出以下改进建议，我将不胜感激：
对sorting.quicksort的调用无效。也许我说的不对。
我想使用spark mllib类multivariatestatisticalsummary来计算统计值。
为此，我需要将所有中间值保留为rdd，以便可以直接使用rdd方法来完成这项工作。
最后，我还需要将结果写入hdfs，rdd类上提供了一个方法来执行此操作，这也是我希望将所有内容保留为rdd的另一个原因。