使用Spark/Scala格式化日誌

Question

我有一個包含一些我想通過Spark處理的信息的日誌文件。唯一的問題是整個文件本身格式不正確。 所以我試圖整齊地格式化，只抓取我需要的數據。

現在我已經注意到大部分有用的信息都包含一個「INFO」標籤。所以，我決定使用以過濾由：

VAL testje = realdata.filter（線=> line.contains（ 「INFO」））

但現在我想將所得數據到SQLContext處理至i可以將數據可視化（在齊柏林飛艇中）然而，

• 由此產生的RDD仍然有很多垃圾我可能不需要。
• 當我嘗試格式化與案例類我總是得到一個ArrayOutofBounds錯誤。可能是因爲垃圾信息比我在課堂上定義的時間長 。

這裏是什麼樣的數據貌似現在的（非常小）例如：

2016-03-08 14:55:29,637 INFO [ajp-nio-8009-exec-1] n.t.f.s.FloorService [FloorService.java:281] Snoozing. Wait 569 more milliseconds. Time passed : 4431 
2016-03-08 14:55:29,964 INFO [ajp-nio-8009-exec-3] n.t.f.c.FloorUpdateController [FloorUpdateController.java:67] Floor test received update from tile: 1, data = [false, false, false, false, false, false, false, false] 
2016-03-08 14:55:30,582 INFO [ajp-nio-8009-exec-2] n.t.f.c.FloorUpdateController [FloorUpdateController.java:67] Floor test received update from tile: 1, data = [false, false, false, false, false, false, true, false] 
2016-03-08 14:55:30,592 INFO [ajp-nio-8009-exec-2] n.t.f.s.FloorService [FloorService.java:284] delta time : 5387 
2016-03-08 14:55:30,595 INFO [ajp-nio-8009-exec-2] n.t.f.s.ActivityService [ActivityService.java:31] Activity added for floor with id: test 
2016-03-08 14:55:30,854 INFO [ajp-nio-8009-exec-4] n.t.f.c.FloorUpdateController [FloorUpdateController.java:67] Floor test received update from tile: 1, data = [false, false, false, false, false, false, false, false] 

我真正需要的是日期，時間，區塊ID和布爾值。

有什麼辦法可以正確格式化，而不必考慮所有的垃圾數據？

這裏就是我想現在（免責聲明，我是相當新的這一點，我有點即興發揮^^「）：

import org.apache.commons.io.IOUtils 
import java.net.URL 
import java.nio.charset.Charset 

val realdata = sc.textFile("/media/application.txt") 

case class testClass(date: String, time: String, level: String, unknown1: String, unknownConsumer: String, unknownConsumer2: String, vloer: String, tegel: String, msg: String, bool1: String, bool2: String, bool3: String, bool4: String, bool5: String, bool6: String, bool7: String, bool8: String, batchsize: String, troepje1: String, troepje2: String) 

//val testje = realdata.filter(line => line.contains("INFO")) 
val mapData = realdata.map(s => s.split(" ")).filter(line => line.contains("INFO")).map(
    s => testClass(s(0), 
     s(1), 
     s(2), 
     s(3), 
     s(4), 
     s(5), 
     s(6), 
     s(7), 
     s(8), 
     s(9), 
     s(10), 
     s(11), 
     s(12), 
     s(13), 
     s(14), 
     s(15), 
     s(16), 
     s(17), 
     s(18), 
     s(19) 
     ) 
    ).toDF() 
    mapData.registerTempTable("test") 

Answer 1

我建議您使用data而不是INFO進行過濾，因爲要分割並轉換爲數據幀的行包含data。
我已經修改了代碼，以滿足您的case class一點點，你可以編輯更根據您的需要

val mapData = realdata 
.filter(line => line.contains("data")) 
.map(s => s.split(" ").toList) 
.map(
    s => testClass(s(0), 
    s(1).split(",")(0), 
    s(1).split(",")(1), 
    s(3), 
    s(4), 
    s(5), 
    s(6), 
    s(7), 
    s(8), 
    s(15), 
    s(16), 
    s(17), 
    s(18), 
    s(19), 
    s(20), 
    s(21), 
    s(22), 
    "", 
    "", 
    "" 
) 
) 
.toDF() 
mapData.show(false) 

希望它可以幫助

Answer 2

我會嘗試做這樣的事情：

val regex = """^(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}:\d{2}),(\d{0,3}) INFO .+Floor test received update from tile: (\d+), data = (\[((false|true)(,){0,1})+\])$""".r 
final case class LogLine(date: Instant, tileId: String, data: Seq[Boolean]) 
realdata.flatMap({ 
    case regex(date, time, millis, tileId, data, _*) => 
    val mapper = new ObjectMapper() with ScalaObjectMapper 
    mapper.registerModule(DefaultScalaModule) 

    Seq(LogLine(
     Instant.parse(s"${date}T$time.${millis}Z"), 
     tileId, 
     mapper.readValue[Seq[Boolean]](data) 
    )) 
    case _ => Nil 
}) 

案例類將是多維的，但這是你可能想要在這種情況下。如果你真的需要它，你可以隨時把它弄平。

如果要提高性能，可以使用mapPartitions而不是flatMap並重新使用ObjectMapper。