在 Spark 中的一个文件中读取多个 Json

Question

我正在尝试在 spark 中读取每行有一个 json 文件的 json 文件

["io", {"in": 8, "out": 0, "dev": "68", "time": 1532035082.614868}]
["io", {"in": 0, "out": 0, "dev": "68", "time": 1532035082.97122}]
["test", {"A": [{"para1":[], "para2": true, "para3": 68, "name":"", "observation":[[2,3],[3,2]],"time": 1532035082.97122}]}]

这有点棘手，因为每一行都是一个有效的 json 文件。 我直接和熊猫一起做：

pd.read_json(filepath,compression='infer', orient='records, lines=True)

但是在使用 DataFrame 的 spark 中它不起作用

 spark.read.option('multiline','true').json(filepath)

我试图逐行读取文件，但仍然出现错误：

   lines = sc.textFile(filepath)
   llist = lines.collect()
   for line in llist:
        print(line)
        df = spark.read.option('multiline','true).json(line)
        df.printSchema()

错误是 IllegalArgumentException： java.net.URISyntaxException：绝对 URI 中的相对路径：.....

感谢您帮助找出解决方案

Answer 1

一种可能的方法是读取文本文件并将每一行解析为两个字符串的数组：

import pyspark.sql.functions as F

df = spark.read.text(filepath).withColumn(
    'value', 
    F.from_json('value', 'array<string>')
).select(
    F.col('value')[0].alias('c0'), 
    F.col('value')[1].alias('c1')
)

df.show(truncate=False)
+----+------------------------------------------------------------------------------------------------------------+
|c0  |c1                                                                                                          |
+----+------------------------------------------------------------------------------------------------------------+
|io  |{"in":8,"out":0,"dev":"68","time":1.532035082614868E9}                                                      |
|io  |{"in":0,"out":0,"dev":"68","time":1.53203508297122E9}                                                       |
|test|{"A":[{"para1":[],"para2":true,"para3":68,"name":"","observation":[[2,3],[3,2]],"time":1.53203508297122E9}]}|
+----+------------------------------------------------------------------------------------------------------------+

但请注意，c1 列是字符串类型。 Spark 不可能像 pandas 一样在列中保存具有不同模式的字典。