Как вычислить совокупную сумму распада в PySpark с помощью рекурсивной операции? ⇐ Python

[Anonymous]

У меня есть кадр данных PySpark, который выглядит следующим образом:



id
id2
id3
h_generated
коэффициент распада
h_total


< tbody>

1
164
1
< td>149.8092121




1
164
2
1417,298433
0,944908987
1558.854504


1
164
3
3833,995761
0,886920437
5216,575679< /td>


1
164
4
285,9751331
0,816006194
4542,733199


1
164
5
309.3110378< /td>
0.926198535
4516.783871


2
315
1
97.6314541




2
315
2
335,8205993
0,881027299
335,8205993

2
315
3
3549.666563
0.735895859
3796.795552


2
315
4
2802.0590060.840857282
5994,622196


2
315
5
2748,337439
0,592542112
6300.403536



Первый заполненный расчет h_total для каждый идентификатор:

Код: Выделить всё

h_total[1] = (h_generated[0] * decay_factor[1]) + h_generated[1]

Расчет последующих строк для каждого идентификатора:

Код: Выделить всё

h_total[n] = (h_total[n-1] * decay_factor[n]) + h_generated[n]

По сути, эта формула накапливает h_total по строкам, принимая во внимание затухание от предыдущих строк.
Что я пробовал
Я попытался использовать цикл for в PySpark для итеративного заполнения значений h_total, но из-за ограничений PySpark с рекурсией этот подход был медленным и неэффективным для моего большого набора данных (миллионы строк).

Код: Выделить всё

from pyspark.sql import Window
import pyspark.sql.functions as F

# Define the cooling constant
cooling_constant = 300

# Define a window to partition by id
window_spec = Window.partitionBy("id")

# Step 1: Initialize h_total for the first row in each partition
test_df = sorted_df.withColumn(
"h_total",
F.when(F.row_number().over(window_spec) == 1, F.col("h_generated"))
)

# Step 2: Iteratively update the h_total for each row
# We will loop a number of times to propagate the h_total value to all rows
for i in range(1, sorted_df.count()):  # Increase the number of iterations as needed for your dataset
test_df = test_df.withColumn(
"h_total",
F.when(F.col("h_total").isNotNull(), F.col("h_total"))
.otherwise(
(F.lag("h_total").over(window_spec) * F.col("decay_factor")) + F.col("h_generated")
)
)

Самая большая проблема, с которой я столкнулся, заключается в том, что я хотел бы избежать использования цикла for, поскольку мой набор данных содержит миллионы строк. Использование цикла таким образом не позволяет мне в полной мере использовать возможности кластера Spark для распределенной обработки.
Есть ли способ решить эту проблему в PySpark, не прибегая к циклу? В идеале мне бы хотелось решение, которое использует преимущества распределенных возможностей Spark для повышения производительности и масштабируемости.

Подробнее здесь: https://stackoverflow.com/questions/791 ... ke-operati