ウィンドウ内で固定されている列の値に基づいてカウンタをインクリメントする方法はありますか？

Question

私は、時間がたつにつれて特定のユーザーがいる地域を示すデータセットを持っています。このデータセットから、私は彼らが各場所で過ごした夜の数を計算したいと思います。 「夜を過ごす」とは、特定の日の23時59分までのユーザーの最後の場所を取ることです。そのユーザーからその翌日の05:00までのすべての観測された場所、またはまだ1つも見つからない場合は、前の日の最後に一致する、その場所で過ごした夜です。

| Timestamp| User| Location| 
|1462838468|49B4361512443A4DA...|1| 
|1462838512|49B4361512443A4DA...|1| 
|1462838389|49B4361512443A4DA...|2| 
|1462838497|49B4361512443A4DA...|3| 
|1465975885|6E9E0581E2A032FD8...|1| 
|1457723815|405C238E25FE0B9E7...|1| 
|1457897289|405C238E25FE0B9E7...|2| 
|1457899229|405C238E25FE0B9E7...|11| 
|1457972626|405C238E25FE0B9E7...|9| 
|1458062553|405C238E25FE0B9E7...|9| 
|1458241825|405C238E25FE0B9E7...|9| 
|1458244457|405C238E25FE0B9E7...|9| 
|1458412513|405C238E25FE0B9E7...|6| 
|1458412292|405C238E25FE0B9E7...|6| 
|1465197963|6E9E0581E2A032FD8...|6| 
|1465202192|6E9E0581E2A032FD8...|6| 
|1465923817|6E9E0581E2A032FD8...|5| 
|1465923766|6E9E0581E2A032FD8...|2| 
|1465923748|6E9E0581E2A032FD8...|2| 
|1465923922|6E9E0581E2A032FD8...|2| 

私はここにウィンドウ関数を使用する必要が推測している、と私は、過去に他のもののためにPySparkを使用しましたが、私はここに開始する場所へと途方に暮れて少しです。私は最後にあなたがイベントや出力の夜のシリーズを取る機能を持っている必要がありますだと思う

Answer 1

は、使用済み...のようなものを（例えば、単にアイデアを得るために）：

def nights_spent(location_events): 
    # location_events is a list of events that have time and location 

    location_events = sort_by_time(location_events) 

    nights = [] 

    prev_event = None 
    for event in location_events[1:]: 
     if prev_location is not None: 
      if next_day(prev_event.time, event.time) \ 
       and same_location(prev_event.location, event.location): 
       # TODO: How do you handle when prev_event 
       # and event are more than 1 day apart? 
       nights.append(prev_location) 

     prev_location = location 

    return nights 

その後、私は最初の良いアプローチは、まずユーザーごとにグループ化して、特定のユーザーのすべてのイベント（場所と時間を含む）を取得することだと考えてください。

次に、イベントのリストを上記の関数に供給すると、すべて(user, nights_spent)行がRDDに格納されます。あなたが始めるのに役立ちます

nights_spent_per_user = all_events.map(lambda x => (x.user, [(x.time, x.location)])).reduce(lambda a, b: a + b).map(x => (x[0], nights_spent(x[1]))) 

希望：

ので、一般的に、RDDは次のようになります。