将跨列值、火花放电和旋转值组合起来

Question

我有一个这样的pypark：

+-------+-------+-----------+-----------+-----------+-----------+-----------+-----------+
| SEQ_ID|TOOL_ID|kurtosis_1m|kurtosis_2m|kurtosis_3m|kurtosis_4m|kurtosis_5m|kurtosis_6m|
+-------+-------+-----------+-----------+-----------+-----------+-----------+-----------+
|3688539|  99725|     6.7484|     6.2753|     6.2055|     7.2076|     7.0501|     7.5099|
|3689076|  99705|     4.8109|     4.3774|     4.1131|     4.4084|     4.1568|     4.4445|
+-------+-------+-----------+-----------+-----------+-----------+-----------+-----------+

我需要以这样的方式来旋转它，这样我就得到了这样的数据数据：

+-------+-------+-----------+
| SEQ_ID|TOOL_ID|kurtosis   |
+-------+-------+-----------+
|3688539|  99725|     6.7484|
|3688539|  99725|     6.2753|
|3688539|  99725|     6.2055|
|3688539|  99725|     7.2076|
|3688539|  99725|     7.0501|
|3688539|  99725|     7.5099|
|3689076|  99705|     4.8109|
|3689076|  99705|     4.3774|
|3689076|  99705|     4.1131|
|3689076|  99705|     4.4084|
|3689076|  99705|     4.1568|
|3689076|  99705|     4.4445|
+-------+-------+-----------+

我认为一种方法是将kurtosis列创建为数组列，然后对其进行爆炸。如何将跨dataframe的列的值作为数组组合到单个列中。我还有其他专栏如mean_1m，mean_2m等..。我需要以同样的方式转向。有什么见解吗？谢谢

Answer 1

您可以使用数组将多个列组合成一个数组，然后--就像您在问题中已经描述过的那样-- 分解数组。

from pyspark.sql import functions as F

cols = [x for x in df.schema.names if "kurtosis" in x]

df.withColumn("kurtosis", F.explode(F.array(cols))) \
    .drop(*cols) \
    .show()

输出：

+-------+-------+--------+
| SEQ_ID|TOOL_ID|kurtosis|
+-------+-------+--------+
|3688539|  99725|  6.7484|
|3688539|  99725|  6.2753|
|3688539|  99725|  6.2055|
|3688539|  99725|  7.2076|
|3688539|  99725|  7.0501|
|3688539|  99725|  7.5099|
|3689076|  99705|  4.8109|
|3689076|  99705|  4.3774|
|3689076|  99705|  4.1131|
|3689076|  99705|  4.4084|
|3689076|  99705|  4.1568|
|3689076|  99705|  4.4445|
+-------+-------+--------+

Answer 2

您可以创建一个数据格式数组，并将它们合并为。

首先，标识kurtosis列：

sub_string = "kurtosis"
kurtosis_col = [x for x in df.schema.names if sub_string in x]

现在，创建一个等于kurtosis列数的数据数组：

from functools import reduce
from pyspark.sql import DataFrame
from pyspark.sql.functions import col

df_array = [df.withColumn('col', F.concat(F.col(x)))
                .select('seq_id', 'tool_id', 'col') for x in kurtosis_col]

# Union them
reduce(DataFrame.unionAll, df_array).withColumnRenamed("col", "kurtosis").show()

输出：

+-------+-------+--------+
| seq_id|tool_id|kurtosis|
+-------+-------+--------+
|3688539|  99725|  6.7484|
|3688539|  99725|  7.2076|
|3688539|  99725|  6.2753|
|3688539|  99725|  6.2055|
|3688539|  99725|  7.5099|
|3688539|  99725|  7.0501|
|3689076|  99705|  4.4084|
|3689076|  99705|  4.1131|
|3689076|  99705|  4.8109|
|3689076|  99705|  4.4445|
|3689076|  99705|  4.3774|
|3689076|  99705|  4.1568|
+-------+-------+--------+

您可以对其他设置列(如mean_1m等)采用类似的方法。将它们连接回并避免重复的一种方法是在加入之前使用row_number()和monotonically_increasing_id()。如果你需要那一段代码，请告诉我。