PySpark：使用数组值作为列标题从嵌套数组动态创建数据框

Question

我遇到了一个问题，希望有人能提供帮助。

假设我有如下记录的传入数据流：

{ “标题”：[“col_a”，“col_b”，“col_c”，“col_d”]，“数据”：[[“0”，“1”，“2”，“3”]，[ "0.2","0.1","3","4"],["5","4","3","2"]]}

{ “标题”：[“col_a”，“col_b”，“col_c”，“col_d”]，“数据”：[[“0.1”，“1.2”，“2.5”，“3”]，[ "0","0","1","0"]]}

...

现在进一步假设数据被清理，使得：

1. “headers”字段始终包含相同的数组
2. “数据”数组中的数组始终与标头数组长度相同

在 PySpark 中是否可以将上述记录转换为如下所示的数据框？

col_a	col_b	col_c	col_d
0	1	2	3
0.2	0.1	3	4
5	4	3	2
0.1	1.2	2.5	3
0	0	1	0

非常感谢任何 cmets 和/或工作代码。

Answer 1

我找到了两种方法

1. 使用地图
2. 使用数据透视操作 - 为此，您需要能够将所有数据收集到分组操作的单个分区中。所以它会在大型数据集上失败

地图解决方案

  df = spark.read.json("map.json")
  
  # collect the headers as a list
  headers = df.select(F.explode("headers").alias("headers")).distinct().collect()
  headers = [r.headers for r in headers]
  
  # explode data arrays so that it has the same dimensions as the header array
  df = df.select(F.explode("data").alias("data"), "headers")
  # zip data and headers together to form a map
  df = df.select(F.arrays_zip("headers", "data").alias("map"))
  df = df.select(F.map_from_entries("map").alias("map"))
  
  #select out your headers from the map to form columns
  df = df.select(*[df.map.getItem(col).alias(col) for col in headers])
  df.show(truncate=False)
  >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
  +-----+-----+-----+-----+
  |col_a|col_b|col_c|col_d|
  +-----+-----+-----+-----+
  |0    |1    |2    |3    |
  |0.2  |0.1  |3    |4    |
  |5    |4    |3    |2    |
  +-----+-----+-----+-----+

枢轴解决方案

  df = spark.read.json("map.json")

  # explode data arrays so that it has the same dimensions as the header array
  df = df.select(F.explode("data").alias("data"), "headers")
  # zip data and headers together and explode it into rows
  df = df.select(F.arrays_zip("headers", "data").alias("zipped"))
  df = df.select(F.explode("zipped").alias("exploded_struct"))
  df = df.selectExpr("exploded_struct.*")
  
  # Add a single index so that we can group by it and then pivot our headers into columns. 
  # **NB This groups all of our data into a single partition
  df = df.withColumn("idx", F.lit(1))
  df = df.groupBy("idx").pivot("headers").agg(F.collect_list("data").alias("data")).drop("idx")
  
  # each column now contains it's array of data. in order to explode it we need to zip all of them 
  # together and explode in a single operation
  df = df.withColumn("zipped",F.arrays_zip(*df.columns))
  df = df.select(F.explode("zipped").alias("exploded_struct"))
  
  # Finally we select out our headers from the exploded_struct
  df = df.selectExpr("exploded_struct.*")
  df.show(truncate=False)

 >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
   +-----+-----+-----+-----+
   |col_a|col_b|col_c|col_d|
   +-----+-----+-----+-----+
   |0    |1    |2    |3    |
   |0.2  |0.1  |3    |4    |
   |5    |4    |3    |2    |
   +-----+-----+-----+-----+