pyspark json 读取以标记不良记录

Question

我想使用 pyspark 解析带有 json 数据的文件，并想标记“坏/意外”记录。 “不良/意外记录”是指那些不遵循我指定的模式的记录。 我有这个输入文件并想指定 schema 。 当数据按照架构的预期格式时，它可以工作。 (inp1.json) 当输入文件中的数据格式不正确时，它不起作用。 (inp2.json) 在这种情况下，它只是将整个文件/数据帧读取为空。 我想要的只是将该一条记录视为损坏的记录并读取其余 3 行。 请有任何建议。

inp1.json  (data in correct format)


[{"last_name": ["ln1", ""], "city": ["c1", "c2"]},
{"last_name": ["ln3", "ln4"], "city": ["c10", "c20"]},
{"last_name": ["ln2"], "city": ["c1", "c2"]}]

from pyspark.sql.types import StructType, StructField, StringType, ArrayType, LongType, DoubleType
myschema = StructType([

     StructField('city', ArrayType(StringType(), True), True),
     StructField('last_name', ArrayType(StringType(), True), True)
 ])

sc = SparkContext(appName=app)


inp_file="inp1.json"
spark = SparkSession.builder.appName("read_json").config("spark.some.config.option","some-value").enableHiveSupport().getOrCreate()
raw_df = spark.read.json(inp_file,multiLine=True, schema=myschema)
print "raw_df"
raw_df.show(truncate=False)


raw_df
+----------+----------+
|city      |last_name |
+----------+----------+
|[c1, c2]  |[ln1, ]   |
|[c10, c20]|[ln3, ln4]|
|[c1, c2]  |[ln2]     |
+----------+----------+

记录不良数据的样本运行

inp2.json  (data in in correct format, please note that last_name in the last record is not an array, but just a string)

[{"last_name": ["ln1", ""], "city": ["c1", "c2"]},
{"last_name": ["ln3", "ln4"], "city": ["c10", "c20"]},
{"last_name": ["ln2"], "city": ["c1", "c2"]},{"last_name": "ln4", "city": ["c4", "c5"]}]


raw_df
+----+---------+
|city|last_name|
+----+---------+
|null|null     |
+----+---------+

Answer 1

读取json时可以指定mode=DROPMALFORMED选项。

raw_df = spark.read.option('mode','DROPMALFORMED').json(inp_file,multiLine=True, schema=myschema)

https://spark.apache.org/docs/2.3.1/api/java/org/apache/spark/sql/DataFrameReader.html#json-scala.collection.Seq-