У меня есть несколько массивов json
[{"key":"country","value":"aaa"},{"key":"region","value":"a"},{"key":"city","value":"a1"}]
[{"key":"city","value":"b"},{"key":"street","value":"1"}]
Мне нужно извлечь значение города и улицы в разные столбцы.
Использование get_json_object($"address", "$[2].value").as("city")
для получения элемента по его номеру не работает, потому что в массивах могут отсутствовать некоторые поля.
Вместо этого я решил превратить этот массив в карту пар ключ -> значение, но у меня возникли проблемы с этим. Пока мне удалось получить только массив массивов.
val schema = ArrayType(StructType(Array(
StructField("key", StringType),
StructField("value", StringType)
)))
from_json($"address", schema)
Возвращает
[[country, aaa],[region, a],[city, a1]]
[[city, b],[street, 1]]
Я не уверен, куда идти отсюда.
val schema = ArrayType(MapType(StringType, StringType))
Сбой с
cannot resolve 'jsontostructs(`address`)' due to data type mismatch: Input schema array<map<string,string>> must be a struct or an array of structs.;;
Я использую искру 2.2
Используя UDF, мы легко справимся с этим. В приведенном ниже коде я создал карту, используя UDF. Я надеюсь, что этого будет достаточно
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._
val df1 = spark.read.format("text").load("file path")
val schema = ArrayType(StructType(Array(
StructField("key", StringType),
StructField("value", StringType)
)))
val arrayToMap = udf[Map[String, String], Seq[Row]] {
array => array.map { case Row(key: String, value: String) => (key, value) }.toMap
}
val dfJSON = df1.withColumn("jsonData",from_json(col("value"),schema))
.select("jsonData").withColumn("address", arrayToMap(col("jsonData")))
.withColumn("city", when(col("address.city").isNotNull, col("address.city")).otherwise(lit(""))).withColumn("street", when(col("address.street").isNotNull, col("address.street")).otherwise(lit("")))
dfJSON.printSchema()
dfJSON.show(false)
не могли бы вы показать входной DF и исключенный DF.