pyspark行转列、列转行或宽表转窄表、窄表转宽表

from pyspark.sql import functions as Fdef unpivot(df, keys):# 参数说明 df  dataframe   keys 待转换表中需要保留的主键key&＃xff0c;以list[]类型传入# 转换是为了避免字段类不匹配&＃xff0c;统一将数据转换为string类型&＃xff0c;如果保证数据类型完全一致&＃xff0c;可以省略该句df &＃61; df.select(*[F.col(_).astype("string") for _ in df.columns])cols &＃61; [_ for _ in df.columns if _ not in keys]stack_str &＃61; &＃39;,&＃39;.join(map(lambda x: "&＃39;%s&＃39;, %s" % (x, x), cols))# feature, value 转换后的列名&＃xff0c;可自定义df &＃61; df.selectExpr(*keys, "stack(%s, %s) as (feature, value)" % (len(cols), stack_str))return df

代码演示

from pyspark.sql import functions as Fdef pivot(df, keys, column, column_value, column_value_list):# 参数说明 df  dataframe   keys 待转换表中需要保留的主键key&＃xff0c;以list[]类型传入 # column 待转换的列名  column_value column对应值的列名  column_value_list column内需要成列值&＃xff0c;也是新生成表的列名# .fillna(-999.0)  行转列时有时对应的列没有值&＃xff0c;就会产生null值&＃xff0c;fillna会对null做处理&＃xff0c;转换为其他值&＃xff0c;如不需要可以删除return df.groupBy(keys).pivot(column, column_value_list).agg(F.first(column_value, ignorenulls&＃61;True)).fillna(-999.0)

代码演示