统计日志数据中的脏数据
现在我们手头上有一个日志文件,里面只有3个字段分别是url,时间,流量。在日志文件里面这些都是字符串,所以无论是时间还是流量都可能是一些不可用的格式,现在我们要统计出有多少脏数据并保留。
先贴代码吧,这里很多东西我都是写死了,大家就当个简单例子来看看吧:
package sparkHomWork
import java.io.{File, PrintWriter}
import java.text.SimpleDateFormat
import org.apache.spark.{SparkConf, SparkContext}
object LogStat {
def main(args: Array[String]): Unit = {
//正常生产上就别这样写啦
val conf = new SparkConf().setMaster("local[2]").setAppName("test")
val sc = new SparkContext(conf)
//读取一个本地日志文件
val file = sc.textFile("logs")
//对每一行的数据进行切分
val lines = file.map(_.split("\t"))
//这里我缓存了一下这个数据,其实我就是手痒写写的,实际上没啥事的
lines.persist()
val num = lines.count()
//我们要统计脏数据的数量,所以我在这里定义了一个计数器
val accum = sc.longAccumulator("Error Accum")
//使用过滤器来将非脏数据过滤掉,并同时累计脏数据条数
val list = lines.filter(x => {
try{
val fm = new SimpleDateFormat("yy-MM-dd HH:mm:ss")
val tm = x(1)
val dt = fm.parse(tm)
val value = x(2).toInt
false
}catch {
case e:Exception => {accum.add(1L);true}
}
})
//这里其实可以将RDD里面的数据处理一下,使用saveAsTextFile这个算子,我就是手痒才
list.map(x => (x(0) + "\t" + x(1) + "\t" + x(2)).saveAsTextFile("errorLogs")
val end = list.collect()
val writer = new PrintWriter(new File("errorNum"))
writer.printer(accum.value)
writer.flush()
writer.close()
println("the number of the error :" + accum.value)
val writer = new PrintWriter(new File("corNum"))
writer.printer(num)
writer.flush()
writer.close()
sc.stop()
}
}
里面的实现过程就是,当我将log数据转成RDD后,先对数据进行切分成3个部分:url,时间,流量。但这个时候都是字符串,我们要将其转数据类型。我把这个过程放到了一个try-catch块里面了,如果在转换的过程中抛出了异常,那么这个数据可能就是脏数据,这个时候catch捕获到这个异常对累加器添加一个位数,同时返回一个true给filter函数,在RDD中保留下来。经过上面这个过程后,我们list这个rdd里面就剩下脏数据了,同时我们也有了脏数据的条数了(从累加器里面拿)。
其实,上面的代码还是可以再优化一点,比如不需要在每个元素的过滤时创建一个new SimpleDateFormat(“yy-MM-dd HH:mm:ss”)对象,我们可以在main方法里面new出来,再用广播变量将这个对象广播到各个exeecutor上面,既然要广播一个对象就要用到序列化,我们也是可以使用kryo序列器来序列化这个对象,但是要记得提前注册好。
|
|