如何将行值（时间序列）转换为 PySpark 中的列值？

Question

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

到目前为止，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     |
+----------+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+------+------+------+------+------+------+------+------+------+------+-------+------+------+------+

目标 DataFrame 应如下所示：

+---------------------+-------------+
| 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 可以做到这一点吗？我已经测试了 pivot 函数，但它不能给我想要的结构。每行应对应一小时。

还有哪些其他的转置可能性？

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")])