使用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。