Как получить следующие строки в файле, если в строке pyspark есть слово ERROR?

Question

У меня есть файл журнала, в котором мне нужно проверять каждую строку.Всякий раз, когда слово «ОШИБКА» появляется в любой строке, мне нужно взять следующие две строки после этой строки.Я должен сделать это в pyspark.

например: Входной файл журнала:

строка 1

строка 2

строка ...ОШИБКА ... 3

строка 4

строка 5

строка 6

Вывод будет:

line 4

line 5

Я создал rdd, используя файл журнала и используя map () для прохождения каждой строки, но я не понимаю точную идею.

Заранее спасибо.

Answer 1

Вот метод с использованием оконных функций:

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# set up DF
df = sc.parallelize([["line1"], ["line2"], ["line3..ERROR"], ["line4"], ["line5"]]).toDF(['col'])

# create an indicator that created a boundary between consecutive errors
win1 = Window.orderBy('col')
df = df.withColumn('hit_error', F.expr("case when col like '%ERROR%' then 1 else 0 end"))
df = df.withColumn('cum_error', F.sum('hit_error').over(win1))

# now count the lines between each error occurrence
win2 = Window.partitionBy('cum_error').orderBy('col')
df = df.withColumn('rownum', F.row_number().over(win2))

# the lines we want are rows 2,3
df.filter("cum_error>0 and rownum in (2,3)").select("col").show(10)```

Answer 2

что-то вроде:

# open your file as f
lines = f.readlines()
for i, line in enumerate(lines):
    if "ERROR" in line:
        print(lines[i+1])
        print(lines[i+2])
        # Exit or something you want to do.