Spark と Scala を使用して Apache アクセス ログを分析する方法

$ sbt/sbt アセンブリ

$ ./bin/spark-shell

scala> val textFile = sc.textFile("README.md") // README.md への参照を作成します scala> textFile.count // このファイルの行数をカウントします scala> textFile.first // 最初の行を出力します

sbt コンパイル
sbt テスト
sbt パッケージ

// これは動作します
$ MASTER=local[4] SPARK_CLASSPATH=AlsApacheLogParser.jar ./bin/spark-shell

// 動作しません
$ MASTER=local[4] ADD_JARS=AlsApacheLogParser.jar ./bin/spark-shell
// 動作しません
spark> :cp AlsApacheLogParser.jar

com.alvinalexander.accesslogparser._ をインポートします。
val p = 新しい AccessLogParser

scala> val log = sc.textFile("accesslog.small")
14/03/09 11:25:23 INFO MemoryStore: curMem=0、maxMem=309225062 で EnsureFreeSpace(32856) が呼び出されました
14/03/09 11:25:23 INFO MemoryStore: ブロックbroadcast_0が値としてメモリに保存されました (推定サイズ 32.1 KB、空き容量 294.9 MB)
ログ: org.apache.spark.rdd.RDD[String] = MappedRDD[1]、textFile、<console>:15
scala> log.count
(ここには大量の出力があります)
res0: ロング = 100000

def getStatusCode(行: Option[AccessLogRecord]) = {
 行一致 {
  ケース Some(l) => l.httpStatusCode
  なしの場合 => "0"
 }
}

log.filter(line => getStatusCode(p.parseRecord(line)) == "404").count

// アクセスログレコードから `request` フィールドを取得します
def getRequest(rawAccessLogString: String): オプション[String] = {
 val accessLogRecordOption = p.parseRecord(rawAccessLogString)
 accessLogRecordOption 一致 {
  ケース Some(rec) => Some(rec.request)
  ケースなし => なし
 }
}

log.filter(line => getStatusCode(p.parseRecord(line)) == "404").map(getRequest(_)).count
val recs = log.filter(line => getStatusCode(p.parseRecord(line)) == "404").map(getRequest(_))
val distinctiveRecs = log.filter(line => getStatusCode(p.parseRecord(line)) == "404").map(getRequest(_)).distinct
個別のRecs.foreach(println)