Spark 中怎么讀取本地日志文件

這篇文章給大家介紹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 中怎么讀取本地日志文件就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。