Options
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
12-01-2021 02:41 AM
Have this finally resolved.
Corrupted rows are flagged with 1 and could be then easly filtered out
#define a schema for col2
from pyspark.sql.types import StructType, StructField
json_schema = ArrayType(StructType([StructField("name", StringType(), nullable = True), StructField("value", StringType(), nullable = True)]))
# from_json is used to validate if col2 has a valid schema. If yes -> correct_json = col2, if no -> correct_json = null
# null is a default value returned by from_json when a valid json could not be created
# rows with corrupted jsons are flagged with 1 by checking a result before and after validation. If col2 was not null and after a validation become null it means that json is corrupted
df = data\
.withColumn("correct_json", from_json(col("col2"), json_schema))\
.withColumn("json_flag", when(col("col2").isNotNull() & col("correct_json").isNull(), 1).otherwise(0))\
.drop("correct_json")
display(df)