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

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