您好,登錄后才能下訂單哦!
這篇文章給大家介紹Spark 中怎么讀取本地日志文件,內(nèi)容非常詳細(xì),感興趣的小伙伴們可以參考借鑒,希望對(duì)大家能有所幫助。
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 = { //設(shè)置本地運(yùn)行,在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,導(dǎo)致 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 } }
關(guān)于Spark 中怎么讀取本地日志文件就分享到這里了,希望以上內(nèi)容可以對(duì)大家有一定的幫助,可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò),可以把它分享出去讓更多的人看到。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。