如何使用可能跨越多行的Spark解析日誌行

Question

我正在開發可讀取和解析自定義日誌文件的Spark/Scala應用程序。我在解析多行日誌條目時遇到問題。以下是我的代碼片段：

case class MLog(dateTime: String, classification: String, serverType: String, identification:String, operation: String) 
val PATTERN = """(?s)(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2},\d{3})\s+(\w+)s+\[(.*)\]\s+\[(.*)\]\s+(.*)""" 


def parseLogLine(log: String): MLog={ 
    val res = PATTERN.findFirstMatchIn(log) 
    if (res.isEmpty) { 
    throw new RuntimeException("Cannot parse log line: " + log) 

    MLog(m.group(1),m.group(2),m.group(3),m.group(4),m.group(5)) 
} 

sc.textFile("/mydirectory/logfile").map(parseLogLine).foreach(println) 

日誌文件中的某些條目跨越多行。正則表達式工作正常單行項，但在一個多行條目閱讀如下所示的一個，

2015-08-31 00:10:17,682 WARN [ScheduledTask-10] [name=custname;mid=9999;ds=anyvalue;] datasource - Scheduled DataSource import failed.     
com.xxx.common.service.ServiceException: system failure: Unable to connect to ANY server: LdapDataSource{id=xxx, type=xxx, enabled=true, name=xxx, host=xxx port=999, connectionType=ssl, username=xxx, folderId=99999} 

我收到此錯誤：

Cannot parse log line:com.xxx.common.service.ServiceException: system failure: Unable to connect to ANY server: LdapDataSource{id=xxx, type=xxx, enabled=true, name=xxx, host=xxx port=999, connectionType=ssl, username=xxx, folderId=99999}

我怎樣才能得到星火閱讀多日誌文件中的行日誌條目？

Answer 1

由於輸入文件很小，您可以使用SparkContext.wholeTextFiles。

// Parse a single file and return all extracted entries 
def parseLogFile(log: String): Iterator[MLog] = { 
    val p: scala.util.matching.Regex = ??? 
    p.findAllMatchIn(log).map(
     m => MLog(m.group(1), m.group(2), m.group(3), m.group(4), m.group(5)) 
    ) 
} 

val rdd: RDD[MLog] = sc 
    .wholeTextFiles("/path/to/input/dir") 
    .flatMap{case (_, txt) => parseLogFile(txt)}