如何在 PySpark 中将行值(时间序列)转置为列值

Question

我有一个Spark DataFrame，我想将其行值转换为单个列。它是时间数据(列 = 小时(。(示例见下文(。

到目前为止，数据帧如下所示：

>>> newdf.show(2)
+----------+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+------+------+------+------+------+------+------+------+------+------+-------+------+------+------+
|Date      |temp0|temp1|temp2|temp3|temp4|temp5|temp6|temp7|temp8|temp9|temp10|temp11|temp12|temp13|temp14|temp15|temp16|temp17|temp18|temp19|temp20 |temp21|temp22|temp23|
+----------+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+------+------+------+------+------+------+------+------+------+------+-------+------+------+------+
|2012-01-07|25   |29   |15   |null |null |null |4    |39   |128  |65   |3     |3     |7     |1     |4     |1     |4     |3     |4     |6     |1      |3     |1     |2     |
|2012-01-08|16   |15   |8    |null |null |null |4    |39   |128  |65   |3     |3     |7     |1     |4     |1     |4     |3     |4     |6     |1      |3     |1     |2     |
+----------+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+------+------+------+------+------+------+------+------+------+------+-------+------+------+------+

目标数据帧应如下所示：

+---------------------+-------------+
| Date                | temperature |
+---------------------+-------------+
| 2012-01-07 00:00:00 | 25          |
| 2012-01-07 01:00:00 | 29          |
| 2012-01-07 02:00:00 | 15          |
| 2012-01-07 03:00:00 | null        |
| ....                | ....        |
| 2012-01-08 00:00:00 | 16          |
| 2012-01-08 01:00:00 | 15          |
+---------------------+-------------+

这在PySpark上可能吗？我已经测试了枢轴函数，但它无法为我提供我想要的结构。每行应对应一个小时。

还有

哪些其他的转调可能性？

Answer 1

步骤 1：创建数据帧，

import pyspark.sql.functions as F
df = sql.createDataFrame([
('2012-01-07',25   ,29   ,15   ,7 ,7 ,7 ,4    ,39   ,128  ,65   ,3     ,3     ,7     ,1     ,4     ,1     ,4     ,3     ,4     ,6     ,1      ,3     ,1     ,2     ),
('2012-01-08',16   ,15   ,8    ,7 ,7 ,7 ,4    ,39   ,128  ,65   ,3     ,3     ,7     ,1     ,4     ,1     ,4     ,3     ,4     ,6     ,1      ,3     ,1     ,2     ),
],[
'Date','temp0','temp1','temp2','temp3','temp4','temp5','temp6','temp7','temp8','temp9','temp10','temp11','temp12','temp13','temp14','temp15','temp16','temp17','temp18','temp19','temp20' ,'temp21','temp22','temp23'
])

步骤 2：分解列并合并以创建时间戳

def _combine(x,y):
    d = str(x) + ' {}:00:00'.format(y)
    return d
combine = F.udf(lambda x,y: _combine(x,y))
cols, dtypes = zip(*((c, t) for (c, t) in df.dtypes if c not in ['Date']))
kvs = F.explode(F.array([
      F.struct(F.lit(c).alias("key"), F.col(c).alias("val")) for c in cols])).alias("kvs")
df = df.select(['Date'] + [kvs])
       .select(['Date'] + ["kvs.key", F.col("kvs.val").alias('temperature')])
       .withColumn('key', F.regexp_replace('key', 'temp', ''))
       .withColumn('Date', combine('Date','key').cast('timestamp'))
       .drop('key')
df.show()

这给出的输出为，

+-------------------+-----------+
|               Date|temperature|
+-------------------+-----------+
|2012-01-07 00:00:00|         25|
|2012-01-07 01:00:00|         29|
|2012-01-07 02:00:00|         15|
|2012-01-07 03:00:00|          7|
|2012-01-07 04:00:00|          7|
|2012-01-07 05:00:00|          7|
|2012-01-07 06:00:00|          4|
|2012-01-07 07:00:00|         39|
|2012-01-07 08:00:00|        128|
|2012-01-07 09:00:00|         65|
|2012-01-07 10:00:00|          3|
|2012-01-07 11:00:00|          3|
|2012-01-07 12:00:00|          7|
|2012-01-07 13:00:00|          1|
|2012-01-07 14:00:00|          4|
|2012-01-07 15:00:00|          1|
|2012-01-07 16:00:00|          4|
|2012-01-07 17:00:00|          3|
|2012-01-07 18:00:00|          4|
|2012-01-07 19:00:00|          6|
|2012-01-07 20:00:00|          1|
|2012-01-07 21:00:00|          3|
|2012-01-07 22:00:00|          1|
|2012-01-07 23:00:00|          2|
|2012-01-08 00:00:00|         16|
|2012-01-08 01:00:00|         15|
|2012-01-08 02:00:00|          8|
|2012-01-08 03:00:00|          7|
|2012-01-08 04:00:00|          7|
|2012-01-08 05:00:00|          7|
|2012-01-08 06:00:00|          4|
|2012-01-08 07:00:00|         39|
|2012-01-08 08:00:00|        128|
|2012-01-08 09:00:00|         65|
|2012-01-08 10:00:00|          3|
|2012-01-08 11:00:00|          3|
|2012-01-08 12:00:00|          7|
|2012-01-08 13:00:00|          1|
|2012-01-08 14:00:00|          4|
|2012-01-08 15:00:00|          1|
|2012-01-08 16:00:00|          4|
|2012-01-08 17:00:00|          3|
|2012-01-08 18:00:00|          4|
|2012-01-08 19:00:00|          6|
|2012-01-08 20:00:00|          1|
|2012-01-08 21:00:00|          3|
|2012-01-08 22:00:00|          1|
|2012-01-08 23:00:00|          2|
+-------------------+-----------+

编辑：在两列的情况下，

cols_1, dtypes = zip(*((c, t) for (c, t) in df.dtypes if 'temp' in c))
cols_2, dtypes = zip(*((c, t) for (c, t) in df.dtypes if 'wind' in c))
kvs = F.explode(F.array([
      F.struct(F.lit(c1).alias("key1"), F.col(c1).alias("val1"), F.lit(c2).alias("key2"), F.col(c2).alias("val2"))
                         for c1,c2 in zip(cols_1,cols_2)
                        ] )).alias("kvs")
df = df.select(['Date'] + [kvs]).select(['Date'] + ["kvs.key1", F.col("kvs.val1").alias('temperature'),
                                                    "kvs.key2", F.col("kvs.val2").alias("wind")])