这篇文章给大家介绍Spark 中怎么读取本地日志文件，内容非常详细，感兴趣的小伙伴们可以参考借鉴，希望对大家能有所帮助。
1.代码如下
import java.io.{FileWriter, BufferedWriter, File}
import com.alvinalexander.accesslogparser.{AccessLogRecord, AccessLogParser}
import org.apache.spark.{SparkContext, SparkConf}
import scala.collection.immutable.ListMap
/**
* Spark 读取本地日志文件,抽取最高的访问地址,排序,并保存到本地文件
* Created by eric on 16/6/29.
*/
object LogAnalysisSparkFile {
def getStatusCode(line: Option[AccessLogRecord]) = {
line match {
case Some(l) => l.httpStatusCode
case None => "0"
}
}
def main(agrs: Array[String]): Unit = {
//设置本地运行,在Vm options:上填写：-Dspark.master=local ，Program arguments上填写：local
val sparkConf = new SparkConf().setMaster("local[1]").setAppName("StreamingTest")
val sc = new SparkContext(sparkConf)
val p = new AccessLogParser
val log = sc.textFile("/var/log/nginx/www.eric.aysaas.com-access.log")
println(log.count())//68591
val log1 = log.filter(line => getStatusCode(p.parseRecord(line)) == "404").count()
println(log1)
val nullObject = AccessLogRecord("", "", "", "", "GET /foo HTTP/1.1", "", "", "", "")
val recs = log.filter(p.parseRecord(_).getOrElse(nullObject).httpStatusCode == "404")
.map(p.parseRecord(_).getOrElse(nullObject).request)
val wordCounts = log.flatMap(line => line.split(" "))
.map(word => (word, 1))
.reduceByKey((a, b) => a + b)
val uriCounts = log.map(p.parseRecord(_).getOrElse(nullObject).request)
.map(_.split(" ")(1))
.map(uri => (uri, 1))
.reduceByKey((a, b) => a + b)
val uriToCount = uriCounts.collect // (/foo, 3), (/bar, 10), (/baz, 1) ...//无序
val uriHitCount = ListMap(uriToCount.toSeq.sortWith(_._2 > _._2):_*) // (/bar, 10), (/foo, 3), (/baz, 1),降序
uriCounts.take(10).foreach(println)
println("**************************")
val logSave = uriHitCount.take(10).foreach(println)
// this is a decent way to print some sample data
uriCounts.takeSample(false, 100, 1000)
//输出保存到本地文件,由于ListMap,导致 saveAsTextFile 不能用
// logSave.saveAsTextFile("UriHitCount")
val file = new File("UriHitCount.out")
val bw = new BufferedWriter(new FileWriter(file))
for {
record <- uriHitCount
val uri = record._1
val count = record._2
} bw.write(s"$count => $uri\n")
bw.close
}
}
2.结果图如下
关于Spark 中怎么读取本地日志文件就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。
