這篇文章給大家介紹Spark 中怎么讀取本地日志文件，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。

成都創(chuàng)新互聯(lián)公司是一家集網(wǎng)站建設(shè),靈壽企業(yè)網(wǎng)站建設(shè),靈壽品牌網(wǎng)站建設(shè),網(wǎng)站定制,靈壽網(wǎng)站建設(shè)報(bào)價(jià),網(wǎng)絡(luò)營(yíng)銷,網(wǎng)絡(luò)優(yōu)化,靈壽網(wǎng)站推廣為一體的創(chuàng)新建站企業(yè)，幫助傳統(tǒng)企業(yè)提升企業(yè)形象加強(qiáng)企業(yè)競(jìng)爭(zhēng)力?？沙浞譂M足這一群體相比中小企業(yè)更為豐富、高端、多元的互聯(lián)網(wǎng)需求。同時(shí)我們時(shí)刻保持專業(yè)、時(shí)尚、前沿，時(shí)刻以成就客戶成長(zhǎng)自我，堅(jiān)持不斷學(xué)習(xí)、思考、沉淀、凈化自己，讓我們?yōu)楦嗟钠髽I(yè)打造出實(shí)用型網(wǎng)站。

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 讀取本地日志文件,抽取最高的訪問(wèn)地址,排序,并保存到本地文件
  * 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) ...//無(wú)序

    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
  }
}

2.結(jié)果圖如下

關(guān)于Spark 中怎么讀取本地日志文件就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。