从json字符串推断模式

Question

我有这个数据框：

cSchema = StructType([StructField("id1", StringType()), StructField("id2", StringType()), StructField("params", StringType())\
                      ,StructField("Col2", IntegerType())])

test_list = [[1, 2, '{"param1": "val1", "param2": "val2"}', 1], [1, 3, '{"param1": "val4", "param2": "val5"}', 3]]

df = spark.createDataFrame(test_list,schema=cSchema) 

+---+---+--------------------+----+
|id1|id2|              params|Col2|
+---+---+--------------------+----+
|  1|  2|{"param1": "val1"...|   1|
|  1|  3|{"param1": "val4"...|   3|
+---+---+--------------------+----+

我要将参数分解为列：

+---+---+----+------+------+
|id1|id2|Col2|param1|param2|
+---+---+----+------+------+
|  1|  2|   1|  val1|  val2|
|  1|  3|   3|  val4|  val5|
+---+---+----+------+------+

所以我把它编码了：

schema2 = StructType([StructField("param1", StringType()), StructField("param2", StringType())])

df.withColumn(
  "params", from_json("params", schema2)
).select(
  col('id1'), col('id2'),col('Col2'), col('params.*')
).show()

问题是params模式是动态的（变量schema2），他可能会从一个执行更改为另一个执行，因此我需要动态地推断模式（所有列都可以是字符串类型）。。。我想不出一个办法

有谁能帮我提一下吗

Answer 1

下面是如何实现的，希望您可以将其更改为python

使用值中的schema_of_json动态获取模式，并使用from_json进行读取

val schema = schema_of_json(df.first().getAs[String]("params"))
df.withColumn("params", from_json($"params", schema))
  .select("id1", "id2", "Col2", "params.*")
  .show(false)

Answer 2

在Pyspark中，语法应为：

import pyspark.sql.functions as F
schema = F.schema_of_json(df.select('params').head()[0])

df2 = df.withColumn(
  "params", F.from_json("params", schema)
).select(
  'id1', 'id2', 'Col2', 'params.*'
)

df2.show()
+ -+ -+  +   +   +
|id1|id2|Col2|param1|param2|
+ -+ -+  +   +   +
|  1|  2|   1|  val1|  val2|
|  1|  3|   3|  val4|  val5|
+ -+ -+  +   +   +