假设你有列id，它决定了行的顺序。
计算group number作为group B出现的运行计数，然后使用collect_list聚合，请参阅代码。它是scala，但相同的spark.sql将在pyspark中工作：

println("Initial data:")
val df1 = Seq(
(1, "XYZ",  "A"),
(2, "PQR" , "A"),
(3,null, "B"   ),
(4,"DEF",   "A"),
(5,"HIJ",   "A"),
(6,"KLM",   "A"),
(7,"NOP",   "A"),
(8,null,    "B"    )
).toDF("Id","Description", "Group")

df1.createOrReplaceTempView("df1")
df1.show(100, false)

println("Result:")
spark.sql(""" 
select 'B' Group, collect_list(Description) Description
from
(
select id, Description, Group, 
       --calculate group number
       count(case when Group='B' then 1 else null end) over(order by id) as grp_num
from df1
) s
group by grp_num
having size(collect_list(Description))>0
order by grp_num

""").show(100, false)

字符串
初始数据：

+---+-----------+-----+
|Id |Description|Group|
+---+-----------+-----+
|1  |XYZ        |A    |
|2  |PQR        |A    |
|3  |null       |B    |
|4  |DEF        |A    |
|5  |HIJ        |A    |
|6  |KLM        |A    |
|7  |NOP        |A    |
|8  |null       |B    |
+---+-----------+-----+

型
测试结果：

+-----+--------------------+
|Group|Description         |
+-----+--------------------+
|B    |[XYZ, PQR]          |
|B    |[DEF, HIJ, KLM, NOP]|
+-----+--------------------+

型