PySpark - 爆炸时访问结构字段名称和值

Question

我的输入数据是这种形式：

[
  {
    "id": 123,
    "embedded": {
      "a": {
        "x": true,
        "y": 1,
      },
      "b": {
        "x": false,
        "y": 2,
      },
    }, 
  },
  {
    "id": 456,
    "embedded": {
      "a": {
        "x": false,
        "y": 3,
      },
      "b": {
        "x": true,
        "y": 4,
      },
    }, 
  },
]

由于我的 pyspark 设置的一些复杂性，embedded 字段是一个 struct，上面有字段 a 和 b。虽然将来可能会有更多 a 和 b 键，但我不想将字段列表硬编码到脚本中。

我希望最终数据的格式为：

_______________________________
| id   |  key   |   x   |   y  |
_______________________________
| 123  |  a     |  true |   1  |
_______________________________
| 123  |  b     | false |   2  |
_______________________________
| 456  |  a     | false |   3  |
_______________________________
| 456  |  b     |  true |   4  |
_______________________________

我知道我想分解 embedded 字段以获取它的不同值，但是如何访问字段名称？

使用下面的 PySpark，我可以提取 id、x 和 y 列的所有值，但是我如何访问结构字段名称（a、b , ...) 什么时候爆炸？

frame = frame.select(
        explode(array("embedded.*")).alias("embedded"),
        frame.id,
    )
    frame = frame.select(
        frame.id,
        frame.embedded.x,
        frame.embedded.y,
    )

Answer 1

如果可能，您应该改用MapType

df = spark_session.createDataFrame([
    Row(id=123, embedded={'a': Row(x=True, y=1), 'b': Row(x=False, y=2)}),
    Row(id=456, embedded={'a': Row(x=False, y=3), 'b': Row(x=True, y=4)})
])
df.select(col("id"), explode(col("embedded"))).select(col("id"), col("key"), col("value.*")).show()

输出：

.+---+---+-----+---+
| id|key|    x|  y|
+---+---+-----+---+
|123|  a| true|  1|
|123|  b|false|  2|
|456|  a|false|  3|
|456|  b| true|  4|
+---+---+-----+---+

StructTypes 与静态模式相关联。如果您真的想为StructTypes 执行此操作，则需要以编程方式生成查询

df = spark_session.createDataFrame([
    Row(id=123, embedded=Row(a=Row(x=True, y=1), b=Row(x=False, y=2))),
    Row(id=456, embedded=Row(a=Row(x=False, y=3), b=Row(x=True, y=4)))
])

field_names = [field.name for field in next(field for field in df.schema.fields if field.name=="embedded").dataType.fields]

dfs = [df.select(col("id"), lit(field_name).alias("key"), col(f"embedded.{field_name}.x"), col(f"embedded.{field_name}.y")) for field_name in field_names]

reduce(lambda x,y: x.union(y), dfs).show()

输出：

+---+---+-----+---+
| id|key|    x|  y|
+---+---+-----+---+
|123|  a| true|  1|
|456|  a|false|  3|
|123|  b|false|  2|
|456|  b| true|  4|
+---+---+-----+---+