PySpark：将列转换为长

在 PySpark 中，可以使用 pivot 和 unpivot 操作来将列转换为长。当我们需要将多列转换为一列时，可以使用 unpivot 操作。反之，当我们需要将一列拆成多列时，可以使用 pivot 操作。

unpivot 操作

以下是使用 PySpark 执行 unpivot 操作的步骤：

1. 固定列：选择要保留的列，这些列不需要进行 unpivot 转换。
2. 转换列：选择要 unpivot 的列，并且为它们设置列名。
3. 堆叠数据：使用 unionAll() 函数合并所有要 unpivot 的列。
4. 提取值：使用 split() 或正则表达式提取值和键名。
5. 删除无用列：删除分离出的包含值和键的列，只留下原始的保留列和添加的新列。

pivot 操作

以下是使用 PySpark 执行 pivot 操作的步骤：

1. 选择要转置的列和值。
2. 使用 groupBy() 函数进行分组。
3. 使用 agg() 函数进行聚合计算。
4. 使用 pivot() 函数进行列转置。
5. 重新安排列的顺序。

示例代码

unpivot 示例代码

from pyspark.sql.functions import split, concat_ws

fixed_cols = ["id", "name"]
unpivot_cols = ["value1", "value2", "value3"]

df = df.select(fixed_cols +
               [concat_ws("_", lit(col), lit(col_name)).alias("key"),
               col(col_name).alias("value") for col_name in unpivot_cols
               for col in [col(col_name)]])
 
df = df.select(fixed_cols + [split(df.key, "_").getItem(1).alias("key_name"), 
                             split(df.key, "_").getItem(2).alias("key_val"), df.value])
 
df = df.drop(df.key)

pivot 示例代码

pivoted = df.groupBy("location").pivot("metric").agg(sum("value"))

pivoted = pivoted.select("location", "metric1", "metric2", "metric3")

以上示例代码是简单的示例，实际情况下需要根据数据的类型和结构进行微调。