PySpark: примените регулярное выражение, чтобы удалить нежелательный текст и сделать ввод допустимым JSON

Question

У меня есть файл JSON, который, к сожалению, содержит нежелательный текст перед каждой строкой:

2019-07-02T22:53:16.848Z LOGFILE {"key":{"host":"example1.net","srcIP":"1.0.0.0","dstIp":"2.0.0.0"},"count":4,"last_seen":"2019-07-02T22:48:15.362Z"}
2019-07-02T22:53:16.937Z LOGFILE {"key":{"host":"example2.net","srcIP":"1.0.0.1","dstIp":"2.0.0.1"},"count":2,"last_seen":"2019-07-02T22:53:07.018Z"}
......

Я хотел бы загрузить этот файл следующим образом:

from pyspark.sql import SparkSession, SQLContext

spark = SparkSession \
       .builder \
       .appName("LogParser") \
       .getOrCreate()

sc = spark.SparkContext()
sqlContext = SQLContext(sc)

df = sqlContext.read.json('log_sample.json')

, но мне нужноспособ удалить этот нежелательный текст, например, 2019-07-02T22:53:16.848Z LOGFILE, чтобы сначала сделать его допустимым JSON.Не могли бы вы пролить свет на то, как применить регулярное выражение, прежде чем я смогу позвонить sqlContext.read.json()?В противном случае он будет жаловаться на это как _corrupt_record.Большое спасибо!

Answer 1

В этом случае вы должны использовать textFile для загрузки всего файла, а затем разбить строку на json, а затем создать из нее DataFrame.

Ниже может быть полезен фрагмент

log_path = './log.txt'

# Load each line to pair rdd
pair_rdd = sc.textFile(log_path)

# Split str from pair rdd and create new rdd of json string
# You can do more thing with it
json_rdd = pair_rdd.map(lambda x: x.split(' LOGFILE ')[1])

# Convert json rdd to DF
original_df = hive_context.read.json(json_rdd)

original_df.printSchema()

Answer 2

Чтение файла с разделителем пробела, удаление первых 2 столбцов и повторное сохранение в формате json.Попробуйте, если это работает для вас.

df=spark.read.csv("file.json",sep=" ").drop("_c0","_c1")

df.write.json("yourjasonfile.json")