如何使用 spark 读取每个日志行以匹配正则表达式模式？答案

【问题标题】：How to read every log line to match a regex pattern using spark?如何使用 spark 读取每个日志行以匹配正则表达式模式？
【发布时间】：2016-12-17 20:32:54
【问题描述】：

下面的程序会报错

from pyparsing import Regex, re
from pyspark import SparkContext
sc = SparkContext("local","hospital")
LOG_PATTERN ='(?P<Case_ID>[^ ;]+);(?P<Event_ID>[^ ;]+);(?P<Date_Time>[^ ;]+);(?P<Activity>[^;]+);(?P<Resource>[^ ;]+);(?P<Costs>[^ ;]+)'
logLine=sc.textFile("C:\TestLogs\Hospital.log").cache()
#logLine='1;35654423;30-12-2010:11.02;register request;Pete;50'
for line in logLine.readlines():
    match = re.search(LOG_PATTERN,logLine)
    Case_ID = match.group(1)
    Event_ID = match.group(2)
    Date_Time = match.group(3)
    Activity = match.group(4)
    Resource = match.group(5)
    Costs = match.group(6)
    print Case_ID
    print Event_ID  
    print Date_Time
    print Activity
    print Resource
    print Costs

错误：

Traceback（最近一次调用最后一次）：文件 “C:/Spark/spark-1.6.1-bin-hadoop2.4/bin/hospital2.py”，第 7 行，在对于 logLine.readlines() 中的行：AttributeError: 'RDD' object has no attribute 'readlines'

如果我添加 open 函数来读取文件，则会收到以下错误：

Traceback（最近一次调用最后一次）：文件 “C:/Spark/spark-1.6.1-bin-hadoop2.4/bin/hospital2.py”，第 7 行，在 f = open(logLine,"r") TypeError: coercing to Unicode: need string or buffer, RDD found

似乎无法弄清楚如何逐行阅读并提取与模式匹配的单词。此外，如果我只通过一个日志行logLine='1;35654423;30-12-2010:11.02;register request;Pete;50' 它可以工作。我是新手，只知道 python 的基础知识。请帮忙。

【问题讨论】：

标签： python apache-spark pyspark

【解决方案1】：

你把事情搞混了。线

logLine=sc.textFile("C:\TestLogs\Hospital.log")

创建一个 RDD，而 RDD 没有 readlines() 方法。在此处查看 RDD API：

http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD

您可以使用 collect() 逐行检索 RDD 的内容。 readlines() 是标准 Python 文件 API 的一部分，但在 Spark 中处理文件时通常不需要它。您只需使用 textFile() 加载文件，然后使用 RDD API 处理它，请参见上面的链接。

【讨论】：

【解决方案2】：

正如 Matei 所回答的，readlines() 是 Python API，sc.textFile 将创建一个 RDD，因此 RDD 没有属性 readlines() 的错误。

如果您必须使用 Spark API 处理文件，您可以在为模式创建的 RDD 上使用过滤器 API，然后您可以根据分隔符拆分输出。

一个例子如下：

    logLine = sc.textFile("C:\TestLogs\Hospital.log")
    logLine_Filtered = logLine.filter(lambda x: "LOG_PATTERN" in x)
    logLine_output  = logLine_Filtered(lambda a: a.split("<delimiter>")[0], a.split("<delimiter>")[1].....).collect()
logLine_output.first()

Dataframe 会更好

【讨论】：