读取json文件时的pyspark损坏记录答案

【问题标题】：pyspark corrupt_record while reading json file读取json文件时的pyspark损坏记录
【发布时间】：2021-05-09 04:56:01
【问题描述】：

我有一个 spark 无法读取的 json(spark.read.json("xxx").show())

{'event_date_utc': None,'deleted': False, 'cost':1 , 'name':'Mike'}

问题似乎是 None 和 False 不在单引号下，spark 不能将它们默认为布尔值、空值甚至字符串。

我试图让我的 spark 读取一个模式，而不是通过强制这 2 列为字符串并具有相同的错误来推断。

对我来说，spark 正在尝试先读取数据，然后应用架构，然后在读取部分失败。

有没有办法告诉 spark 在不修改输入数据的情况下读取这些值？我正在使用 python。

【问题讨论】：

标签： python json apache-spark pyspark

【解决方案1】：

您的输入不是有效的 JSON，因此您无法使用 spark.read.json 读取它。相反，您可以使用 spark.read.text 将其作为文本 DataFrame 加载，并使用 UDF 将 stringified dict 解析为 json：

import ast
import json
from pyspark.sql import functions as F
from pyspark.sql.types import *

schema = StructType([
    StructField("event_date_utc", StringType(), True),
    StructField("deleted", BooleanType(), True),
    StructField("cost", IntegerType(), True),
    StructField("name", StringType(), True)
])

dict_to_json = F.udf(lambda x: json.dumps(ast.literal_eval(x)))

df = spark.read.text("xxx") \
    .withColumn("value", F.from_json(dict_to_json("value"), schema)) \
    .select("value.*")

df.show()

#+--------------+-------+----+----+
#|event_date_utc|deleted|cost|name|
#+--------------+-------+----+----+
#|null          |false  |1   |Mike|
#+--------------+-------+----+----+

【讨论】：

【解决方案2】：

JSON 看起来不太好。需要引用字段值。

您可以先eval 行，它们看起来像是 Python dict 格式。

df = spark.createDataFrame(
    sc.textFile('true.json').map(eval),
    'event_date_utc boolean, deleted boolean, cost int, name string'
)

df.show()
+--------------+-------+----+----+
|event_date_utc|deleted|cost|name|
+--------------+-------+----+----+
|          null|  false|   1|Mike|
+--------------+-------+----+----+

【讨论】：