是否可以采用单个数据框行并将其拆分为多个数据框行？

Question

我是 pySpark 的新手，我正在尝试使用我可以访问的电表间隔数据集 (csv) 中的一些电表数据。

我有一个从 CSV 导入的电表数据创建的数据框架构，看起来像这样：

root
 |-- _c0: string (nullable = true)
 |-- _c1: integer (nullable = true)
 |-- _c2: string (nullable = true)
 |-- _c3: string (nullable = true)
 |-- _c4: integer (nullable = true)
 |-- _c5: string (nullable = true)
 |-- _c6: long (nullable = true)
 |-- _c7: string (nullable = true)
 |-- _c8: string (nullable = true)
 |-- _c9: string (nullable = true)
 |-- _c10: string (nullable = true)
 |-- _c11: double (nullable = true)
 |-- _c12: integer (nullable = true)
 |-- _c13: integer (nullable = true)
 |-- _c14: long (nullable = true)
 |-- _c15: string (nullable = true)
 |-- _c16: double (nullable = true)
 |-- _c17: long (nullable = true)
 |-- _c18: string (nullable = true)
 |-- _c19: double (nullable = true)
 |-- _c20: long (nullable = true)
 |-- _c21: string (nullable = true)
 |-- _c22: double (nullable = true)
 |-- _c23: long (nullable = true)
 |-- _c24: string (nullable = true)
 |-- _c25: double (nullable = true)
 |-- _c26: long (nullable = true)
 |-- _c27: string (nullable = true)
 |-- _c28: double (nullable = true)
 ...

_c13 包含一个数字，表示它后面有多少个 3 数据列的分组。 （表示时间戳、标志和值）每条记录从 1 到 96 不等

(i.e.    timestamp, flag, value)
_c14, _c15, _c16 = Group 1   (202010101315, "NONE", 1.1)
_c17, _c18, _c19 = Group 2   (202010101330, "NONE", 1.2)
_c20, _c21, _c22 = Group 3   (202010101345, "EST", 0.75) etc...

我认为我想要的 AWS Redshift 表的最终输出是源数据帧中每个分组的单行 _c0 到 _c12 将随每个分组输出

+-----+----+     +------+-------------+------+-------+
  _c0   _c1  ... | _c12 | timestamp   | flag | value |
+-----+----+     +------+-------------+------+-------+
   A     B    ...   L    202010101315  NONE    1.1
   A     B    ...   L    202010101330  NONE    1.2
   A     B    ...   L    202010101345  EST     0.75
   etc...

到目前为止，我已经设法将我的数据加载到数据框中。为了遍历每一行，我意识到我可以创建一个带有自定义函数的 RDD 来对行执行操作：

rdd = df.rdd.map(customFunction)

但我很快意识到我只能将单个分组返回给 RDD

然后，我查看了从 customFunction 中将一行附加到新数据帧，但在读取数据帧是不可变的并且每次附加都会返回一个新数据帧之后，我意识到这可能效率不高。

任何关于实现我正在寻找的有效记录拆分的基本结构的帮助将不胜感激！

Answer 1

基本上，您正在寻找explode 函数，该函数允许从一行中的数组创建多行。您还需要从所需的列创建该数组。

但是，我不知道在 sparkSQL 中有什么方法可以创建一个由可变数量的列组成的数组，其大小由另一列参数化。在 RDD 中，您可以轻松地做到这一点并使用flatMap。在 SparkSQL 中，我们可以做的是创建一个包含所有可能列的数组，然后过滤掉你不想要的那些。

假设您的数据框的列列表以您要保留的列开始，然后是包含组数的列（在您的情况下为_c13），然后是包含组的列。在pyspark 中，代码可能如下所示：

# the index of the column containing the number of groups
group_column = 13
# the maximum number of groups, 96 if I understand correctly
max_group_count = 96

df\
  .withColumn("groups", 
       F.array([
           F.struct(
                F.col("_c"+str(group_column+1+3*i)).alias("timestamp"),
                F.col("_c"+str(group_column+2+3*i)).alias("flag"), 
                F.col("_c"+str(group_column+3+3*i)).alias("value"), 
                F.lit(i).alias("index")
           ) for i in range(max_group_count)
       ])
  )\
  .drop(*[
     "_c"+str(i) for i in range(group_column+1, group_column+3*max_group_count+1)
  ])\
  .withColumn("s", F.explode("groups"))\
  .where(F.col("_c"+str(group_column)) > F.col("s.index"))\
  .select([F.col("_c"+str(i)) for i in range(group_column+1)] + ["s.*"])\
  .drop("index")

代码详解：

groups 列是我所说的数组。它包含所有可能的组，每个组都包含在一个结构中，其索引范围从 0 到最大组数。然后我删除组列以避免爆炸。然后我分解数组，每组创建一条线。然后，where 命令使用我们添加到结构中的索引和列_c13 中的组数删除您不想要的组。最后，我们将所有内容都设置为正确的形状：选择我们要保留的列（具有组数的列之前的列）并使用通配符 * 从结构中提取列。