有没有办法展平任意嵌套的 Spark Dataframe？我看到的大部分工作都是为特定模式编写的，我希望能够用不同的嵌套类型（例如 StructType、ArrayType、MapType 等）通用地展平 Dataframe。
假设我有一个类似的架构：
StructType(List(StructField(field1,...), StructField(field2,...), ArrayType(StructType(List(StructField(nested_field1,...), StructField(nested_field2,...)),nested_array,...)))
希望将其调整为具有以下结构的平面表：
field1
field2
nested_array.nested_field1
nested_array.nested_field2
仅供参考，寻找有关 Pyspark 的建议，但也赞赏其他风格的 Spark。
这个问题可能有点老了，但对于那些仍在寻找解决方案的人来说，您可以使用 select * 内联扁平化复杂的数据类型：
首先让我们创建嵌套数据框：
from pyspark.sql import HiveContext
hc = HiveContext(sc)
nested_df = hc.read.json(sc.parallelize(["""
{
"field1": 1,
"field2": 2,
"nested_array":{
"nested_field1": 3,
"nested_field2": 4
}
}
"""]))
现在将其展平：
flat_df = nested_df.select("field1", "field2", "nested_array.*")
您将在此处找到有用的示例： https ://docs.databricks.com/delta/data-transformation/complex-types.html
如果嵌套数组太多，可以使用：
flat_cols = [c[0] for c in nested_df.dtypes if c[1][:6] != 'struct']
nested_cols = [c[0] for c in nested_df.dtypes if c[1][:6] == 'struct']
flat_df = nested_df.select(*flat_cols, *[c + ".*" for c in nested_cols])
这将同时具有结构类型和数组类型的嵌套 df 展平。通过 Json 读取数据时通常会有所帮助。在此https://stackoverflow.com/a/56533459/7131019上进行了改进
from pyspark.sql.types import *
from pyspark.sql import functions as f
def flatten_structs(nested_df):
stack = [((), nested_df)]
columns = []
while len(stack) > 0:
parents, df = stack.pop()
array_cols = [
c[0]
for c in df.dtypes
if c[1][:5] == "array"
]
flat_cols = [
f.col(".".join(parents + (c[0],))).alias("_".join(parents + (c[0],)))
for c in df.dtypes
if c[1][:6] != "struct"
]
nested_cols = [
c[0]
for c in df.dtypes
if c[1][:6] == "struct"
]
columns.extend(flat_cols)
for nested_col in nested_cols:
projected_df = df.select(nested_col + ".*")
stack.append((parents + (nested_col,), projected_df))
return nested_df.select(columns)
def flatten_array_struct_df(df):
array_cols = [
c[0]
for c in df.dtypes
if c[1][:5] == "array"
]
while len(array_cols) > 0:
for array_col in array_cols:
cols_to_select = [x for x in df.columns if x != array_col ]
df = df.withColumn(array_col, f.explode(f.col(array_col)))
df = flatten_structs(df)
array_cols = [
c[0]
for c in df.dtypes
if c[1][:5] == "array"
]
return df
flat_df = flatten_array_struct_df(df)
**
这是我的最终方法：
1）将数据框中的行映射到字典的rdd。在线查找合适的 python 代码来扁平化 dict。
flat_rdd = nested_df.map(lambda x : flatten(x))
在哪里
def flatten(x):
x_dict = x.asDict()
...some flattening code...
return x_dict
2) 将 RDD[dict] 转换回数据帧
flat_df = sqlContext.createDataFrame(flat_rdd)
以下要点将扁平化嵌套 json 的结构，
import typing as T
import cytoolz.curried as tz
import pyspark
def schema_to_columns(schema: pyspark.sql.types.StructType) -> T.List[T.List[str]]:
"""
Produce a flat list of column specs from a possibly nested DataFrame schema
"""
columns = list()
def helper(schm: pyspark.sql.types.StructType, prefix: list = None):
if prefix is None:
prefix = list()
for item in schm.fields:
if isinstance(item.dataType, pyspark.sql.types.StructType):
helper(item.dataType, prefix + [item.name])
else:
columns.append(prefix + [item.name])
helper(schema)
return columns
def flatten_frame(frame: pyspark.sql.DataFrame) -> pyspark.sql.DataFrame:
aliased_columns = list()
for col_spec in schema_to_columns(frame.schema):
c = tz.get_in(col_spec, frame)
if len(col_spec) == 1:
aliased_columns.append(c)
else:
aliased_columns.append(c.alias(':'.join(col_spec)))
return frame.select(aliased_columns)
然后，您可以将嵌套数据展平为
flatten_data = flatten_frame(nested_df)
这将为您提供展平的数据框。
要点取自https://gist.github.com/DGrady/b7e7ff3a80d7ee16b168eb84603f5599