scala dataframe中的collect_list，它将在固定列号的间隔内收集行

Question

我要求将特定分区的所有行收集到DataFrame中的单个行中。我必须将此数据frame转储到cosmosDB中，每个文档只能容纳2MB的数据。但是当我将上面的数据集收集到一行中时，它超过了2MB，并且在写入CosmosDB时抛出了错误。

我想把行收集成一个固定的500行间隔。对于一个分区，前500行应该被收集到一行中，后500行应该被收集到另一行中，依此类推。

输入数据如下所示。

+------+----------+---------------------------------------+
|ID    |TIME      |SGNL                                   |
+------+----------+---------------------------------------+
|00001 |1574360355|{"SN":"Acc","ST":1574360296,"SV":"0.0"}|
|00001 |1574360355|{"SN":"Acc","ST":1574360296,"SV":"0.0"}|
|00001 |1574360355|{"SN":"Acc","ST":1574360296,"SV":"0.0"}|
|00001 |1574360355|{"SN":"Acc","ST":1574360297,"SV":"0.0"}|
|00001 |1574360355|{"SN":"Acc","ST":1574360297,"SV":"0.0"}|
|00001 |1574360355|{"SN":"Acc","ST":1574360297,"SV":"0.0"}|
|00001 |1574360355|{"SN":"Acc","ST":1574360298,"SV":"0.0"}|
+------+----------+---------------------------------------+

我尝试了如下所示，但有些行的大小超过了2MB，无法写入CosmosDB：

val newDF = df.groupBy($"ID", $"TIME").agg(collect_list($"SGNL").as("SGNL"))

+------+----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------+
|ID    |TIME      |SGNL                                                                                                                                                           |
+------+----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------+
|00001 |1574360355|{"SN":"Acc","ST":1574360296,"SV":"0.0"},{"SN":"Acc","ST":1574360296,"SV":"0.0"},{"SN":"Acc","ST":1574360296,"SV":"0.0"},.......................................|
|00002 |1574360355|{"SN":"Acc","ST":1574360297,"SV":"0.0"},{"SN":"Acc","ST":1574360297,"SV":"0.0"},{"SN":"Acc","ST":1574360298,"SV":"0.0"},.......................................| 
|00003 |1574360355|{"SN":"Acc","ST":1574360297,"SV":"0.0"},{"SN":"Acc","ST":1574360297,"SV":"0.0"},{"SN":"Acc","ST":1574360298,"SV":"0.0"},.......................................| 
|00004 |1574360355|{"SN":"Acc","ST":1574360297,"SV":"0.0"},{"SN":"Acc","ST":1574360297,"SV":"0.0"},{"SN":"Acc","ST":1574360298,"SV":"0.0"},.......................................|                                        |
+------+----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------+

+------+----------+--------------------------------------------+
|ID    |TIME      |SGNL                                        |
+------+----------+--------------------------------------------+
|00001 |1574360355|{1st ROW},{2nd ROW},......{500th ROW}       |
|00001 |1574360355|{501st ROW},{502nd ROW},......{1000th ROW}  |
|00001 |1574360355|{1001st ROW},{1002nd ROW},......{1500th ROW}|
|00001 |1574360355|{1501st ROW},{1502nd ROW},......{2000th ROW}|
|..............................................................|
|..............................................................|

|00002 |1574360355|{1st ROW},{2nd ROW},......{500th ROW}       |
|00002 |1574360355|{501st ROW},{502nd ROW},......{1000th ROW}  |
|00002 |1574360355|{1001st ROW},{1002nd ROW},......{1500th ROW}|
|00002 |1574360355|{1501st ROW},{1502nd ROW},......{2000th ROW}|
|..............................................................|
+------+----------+---------------------------------------------+

Answer 1

您可以使用monitonicallyincreasingid构建索引，将其除以500并按该索引进行分组。它将把前500行放在一起，下500行放在一起，以此类推。

df.withColumn("fancy_id", floor(monotonicallyIncreasingId / 500))
  .groupBy("fancy_id")
  .agg(collect_list($"SGNL").as("SGNL"))
  .drop("fancy_id") // if you want to get rid of the artificial id.

如果不想混合ID列，可以使用groupby（“id”,“fancy_id”）。

然而，每个ID的第一组不一定是500。例如，您将得到如下内容:(id1,500元素）、(id1,320元素）、(id2,180元素）、(id2,500元素）、(id2,500元素）、(id2,50元素）、(id3,450元素）...

如果您更喜欢使用(id1,500元素）、(id1,320元素）、(id2,500元素）、(id2,500元素）、(id2,10元素）、(id3,500元素）、(id3,5元素）...，其中第一组总是有500个元素，则可以使用一个窗口：

val w = Window.partitionBy('ID).orderBy('fancy_id)
df.withColumn("fancy_id", monotonicallyIncreasingId)
  .withColumn("rank", rank() over w)
  .groupBy($"ID", floor($"rank" / 500))
  .agg(collect_list($"SGNL").as("SGNL"))