Преобразование функции Python в Spark (scala), обработка временного значения.коллекция

Question

Я нахожусь в процессе преобразования функции Python для подсчета дождя

def extract_cycles(series):
    """
    Returns two lists: the first one containig full cycles and the second
    containing one-half cycles. The cycles are extracted from the iterable
    *series* according to section 5.4.4 in ASTM E1049 (2011).
    """
    points = deque()
    full, half = [], []

    for x in reversals(series):

        points.append(x)

        while len(points) >= 3:
            # Form ranges X and Y from the three most recent points
            X = abs(points[-2] - points[-1])
            Y = abs(points[-3] - points[-2])

            if X < Y:
                # Read the next point
                break

            elif len(points) == 3:
                # Y contains the starting point
                # Count Y as one-half cycle and discard the first point
                half.append(Y)
                points.popleft()

            else:
                # Count Y as one cycle and discard the peak and the valley of Y
                full.append(Y)

                last = points.pop()

                points.pop()

                points.pop()

                points.append(last)

        else:
            # Count the remaining ranges as one-half cycles
            while len(points) > 1:
                half.append(abs(points[-2] - points[-1]))
                points.pop()
    return full, half

Однако я изо всех сил пытаюсь сделать эту работу в стиле Spark - сначала я подумал, что использование window будетработать, но нет способа сохранить промежуточную сумму, на которую может ссылаться следующая строка.

Есть ли другой подход, который я должен рассмотреть?Кажется, что итерации по строкам - мой единственный подход, но он побеждает цель Spark.