如何使用spark读取每个日志行以匹配regex模式？

Question

下面的程序抛出一个错误

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 (most recent call last): File "C:/Spark/spark-1.6.1-bin-hadoop2.4/bin/hospital2.py", line 7, in for line in logLine.readlines(): AttributeError: 'RDD' object has no attribute 'readlines'

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

Traceback (most recent call last): File "C:/Spark/spark-1.6.1-bin-hadoop2.4/bin/hospital2.py", line 7, in 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'，它就可以工作了。我是spark新手，只知道python的基础知识。请帮忙。在

Answer 1

正如Matei所回答的，readlines（）是Python API和sc.TEXT文件将创建RDD，因此RDD没有属性readlines（）的错误。在

如果必须使用sparkapi处理文件，可以在为pattern创建的RDD上使用filter 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()

数据帧会更好

Answer 2

你把事情搞混了。 线

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处理它，请参阅上面的链接。在