pyspark, сравните две строки в кадре данных

Да, вы неправильно используете функцию map. map одновременно работает с одним элементом. Вы можете попробовать использовать оконные функции следующим образом:

from pyspark.sql.functions import col, lag
from pyspark.sql.window import Window

df = (
    sc.parallelize([
        (134, 30, "2016-07-02 12:01:40"), (134, 32, "2016-07-02 12:21:23"),
        (125, 30, "2016-07-02 13:22:56"), (125, 32, "2016-07-02 13:27:07"),
    ]).toDF(["itemid", "eventid", "timestamp"])
    .withColumn("timestamp", col("timestamp").cast("timestamp"))
)

w = Window.partitionBy("itemid").orderBy("timestamp")

diff = col("timestamp").cast("long") - lag("timestamp", 1).over(w).cast("long")

df.withColumn("diff", diff)

Комментарий @ShuaiYuan к исходному ответу правильный. За последний год я намного лучше понял, как работает Spark, и фактически переписал программу, над которой работал для этого поста.

НОВЫЙ ОТВЕТ (27 марта 2017 г.)
Чтобы выполнить сравнение двух строк фрейма данных, я использовал RDD. Я группирую данные по ключу (в данном случае по идентификатору элемента) и игнорирую идентификатор события, поскольку он не имеет значения в этом уравнении. Затем я сопоставляю лямбда-функцию со строками, возвращая кортеж ключа и список кортежей, содержащих начало и конец промежутков событий, который получен из функции «findGaps», которая выполняет итерацию по списку связанных значений (отсортированных временных меток). к каждому ключу. Как только это будет завершено, я отфильтрую ключи без временных промежутков, а затем flatMapValues, чтобы вернуть данные в формате, более похожем на sql. Это делается с помощью следующего кода:

# Find time gaps in list of datetimes where firings are longer than given duration.  
def findGaps(dates, duration):
    result = []
    length = len(dates)

    # convert to dates for comparison
    first = toDate(dates[0])
    last = toDate(dates[length - 1])
    for index, item in enumerate(dates):
        if index < length -1 and (dates[index + 1] - item).total_seconds() > duration:
            # build outage tuple and append to list
            # format (start, stop, duration)
            result.append(formatResult(item, dates[index + 1], kind))
    return result

outage_list = outage_join_df.rdd\
                            .groupByKey()\
                            .map(lambda row: (
                                     row[0],
                                     findGaps(
                                         sorted(list(row[1])), 
                                         limit
                                     )
                                  )
                            )\
                            .filter(lambda row: len(row[1]) > 0)\
                            .flatMapValues(lambda row: row)\
                            .map(lambda row: (
                                 row[0]['itemid'],     # itemid
                                 row[1][0].date(),     # date
                                 row[1][0],            # start
                                 row[1][1],            # stop
                                 row[1][2]             # duration
                            ))\
                            .collect()

ИСХОДНЫЙ ОТВЕТ (НЕВЕРНЫЙ)
Мне удалось решить эту проблему с помощью mapPartitions:

def findOutage(items):
    outages = []

    lastStamp = None
    for item in items:
        if lastStamp and (lastStamp - item.stamp).total_seconds() > 14400:
            outages.append({"item": item.itemid, 
                            "start": item.stamp.isoformat(),
                            "stop": lastStamp.isoformat()})
        lastStamp = item.stamp
    return iter(outages)

items = df.limit(10).orderBy('itemid', desc('stamp'))

outages = items.mapPartitions(findOutage).collect()

Спасибо всем за помощь!