Как объединить две таблицы с помощью агрегации

Используйте искровой SQL. Я использовал Scala, но SQL в pyspark точно такой же, и я считаю, что вы можете легко преобразовать его, если есть какие-либо различия для pyspark.

Объедините два кадра данных, используйте Collect_list(), чтобы получить массив результатов, затем используйте concat_ws(), чтобы объединить массив в строку:

val dF1 = Seq(
("bob", 1, 3),
("john",  5, 8)
).toDF("name","start","end")

dF1.createOrReplaceTempView("dF1")

val dF2 = Seq(
(1, "a"),
(2, "c"),
(3, "d"),
(4, "a"),
(5, "e"),
(6, "c"),
(7, "u"),
(8, "l")
).toDF("day","outcome")

dF2.createOrReplaceTempView("dF2")


spark.sql(""" 
select d1.name, concat_ws('',collect_list(d2.outcome)) outcome
from
(select d1.name, e.day 
  from dF1 d1 
       lateral view explode(sequence(d1.start, d1.end)) e as day
)d1
left join dF2 d2 on d1.day=d2.day
group by d1.name
""").show(100, false)

Результат:

+----+-------+
|name|outcome|
+----+-------+
|bob |acd    |
|john|ecul   |
+----+-------+

Исправление ООМ:

spark.sql(""" 
select d1.name, concat_ws('',collect_list(d2.outcome)) outcome
from dF1 d1 
left join dF2 d2 on d1.start<=d2.day and  d1.end>=d2.day
group by d1.name
""").show(100, false)

Код

df = df1.join(df2, on=(df1.start <= df2.day) & (df1.end >= df2.day))

df = df.groupBy('name').agg(F.collect_list(F.struct('day', 'outcome')).alias('outcome'))
df = df.withColumn('outcome', F.transform(F.array_sort('outcome'), lambda x: x.outcome))
df = df.withColumn('outcome', F.array_join('outcome', ''))

Как это работает?

Используйте метод неравноправного соединения, чтобы соединить фрейм данных df1 с df2

+----+-----+---+---+-------+
|name|start|end|day|outcome|
+----+-----+---+---+-------+
|bob |1    |3  |1  |a      |
|bob |1    |3  |2  |c      |
|bob |1    |3  |3  |d      |
|john|5    |8  |5  |e      |
|john|5    |8  |6  |c      |
|john|5    |8  |7  |u      |
|john|5    |8  |8  |l      |
+----+-----+---+---+-------+

Сгруппируйте фрейм данных по name и collect парам дня и результата. Этот шаг имеет решающее значение, поскольку столбец day необходим для поддержания порядка конкатенации.

+----+--------------------------------+
|name|outcome                         |
+----+--------------------------------+
|john|[{5, e}, {6, c}, {7, u}, {8, l}]|
|bob |[{1, a}, {2, c}, {3, d}]        |
+----+--------------------------------+

Отсортируйте массив пар и преобразуйте его, чтобы извлечь значения outcome.

+----+------------+
|name|outcome     |
+----+------------+
|john|[e, c, u, l]|
|bob |[a, c, d]   |
+----+------------+

Присоединитесь к массиву, чтобы получить результат

+----+-------+
|name|outcome|
+----+-------+
|john|ecul   |
|bob |acd    |
+----+-------+