在 Spark Dataframe 中提取解码列的特定结构字段

Question

我在包含 JSON 数据的 spark 数据框中有一个解码列。例如，

第 1 行：{'id': 1, 'name': 'Slatan', 'surname': 'Yav', 'age': 24}
第 2 行：{'id': 2, 'name': 'Nikos', 'surname': 'Stef', 'age': 34}
第 3 行：{'id': 3, 'name': 'Panos', 'surname': 'Rodes', 'age': 44}

我只想提取“姓名”、“姓氏”、“年龄”并将它们添加到 Spark 数据框中。

我的方法

schema = StructType(
    [
      StructField('name', StringType(), True),
      StructField('surname', StringType(), True),
      StructField('age', IntegerType(), True)
    ]
)

final_df=None
final_df = spark.createDataFrame([], schema)
columns_selected=schema.fieldNames()

final_df=final_df.union(avro_decoded.withColumn("Body_decoded", from_json("Body_decoded", schema)).select(col( 'Body_decoded.*columns_selected')))

但我得到了错误。如何仅使用名称、姓氏和排除 id 值字段填充 final_df（spark 数据框）？

错误

'No such struct field *columns_selected in name, surname;'

解码前的正文列

avro_decoded=read_avro_data.withColumn('Body_decoded', decode(read_avro_data.Body, charset="UTF-8")).select("Body_decoded")

avro_decoded.printSchema()

返回：

root
 |-- Body_decoded: string (nullable = true)

比你提前任何建议。

Answer 1

您在尝试使用 from_json 时采用了正确的方法，但是，您的 json 格式不适合 spark， 因为您有一个 @987654322 @ 因此，只需使用 regexp_replace 将整数用引号括起来，然后使用 from_json 到 select 您的所需的列，如下所示。

df.show() #sample dataframe

#+---------------------------------------------------------+
#|Body_decoded                                             |
#+---------------------------------------------------------+
#|{'id': 1, 'name': 'Slatan', 'surname': 'Yav', 'age': 24} |
#|{'id': 2, 'name': 'Nikos', 'surname': 'Stef', 'age': 34} |
#|{'id': 3, 'name': 'Panos', 'surname': 'Rodes', 'age': 44}|
#+---------------------------------------------------------+
#root
 #|-- Body_decoded: string (nullable = true) #schema


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

schema=MapType(StringType(),StringType())
columns_selected=['name','surname']

df.withColumn("Body_decoded", F.from_json(F.regexp_replace("Body_decoded",'[^"\d,]?(\d+)',"'$1'")\
                                              ,schema))\
  .select(*["Body_decoded.{}".format(x) for x in columns_selected]).show()

#+------+-------+
#|name  |surname|
#+------+-------+
#|Slatan|Yav    |
#|Nikos |Stef   |
#|Panos |Rodes  |
#+------+-------+