Sollte ich die Zustandsberechnung verwenden? Spark Streaming-Status Berechnung Erklärung

Question

Hier ist mein Fall: Ich erhalte Daten von verschiedenen Geräten, die ihre eigene Unterschrift, einen Zeitstempel und ein Flag haben. I-Filter dann die (flag==SAVE_VALUE) in einer Datei, die eine foreachRDD Funktion verwenden, aber nur, wenn es geht um diesen Zustand:

(it is the first time I receive this signature) 
OR 
(I already have this signature && the timestamp is older than an hour) 

Dies, bis ich in einer lokalen Umgebung war, bedeutete für mich eine Karte zu verwenden, wo ich alle gespeicherten die IDs und der letzte empfangene Zeitstempel. Gerade jetzt möchte ich diese Logik in einem Spark wie einen bewegen. Wie soll ich das machen?
Ich fühle mich dies ein Fall für eine Stateful DSTREAM ist, aber ich kann nicht ganz verstehen:

• Wie soll ich RDD in einem DSTREAM eine kartenähnliche speichern? Oder wie erstelle ich eine einzige "Karte RDD"
• Wie kann ich die neuen Daten vergleichen?

Answer 1

Werfen Sie einen Blick auf mapWithState(), es ist genau das, was Sie wollen.

In der StateSpecFunction können Sie bestimmen, ob Sie den aktuellen Status aktualisieren, beibehalten oder entfernen möchten, wenn für denselben Schlüssel ein neuer Wert eintrifft. Sie haben Zugriff auf den aktuellen Status und den neuen Zustand, sodass Sie jede Art von Vergleich zwischen beiden durchführen können.

Es hat auch eine integrierte Unterstützung für Timeouts und kann auf mehrere Executoren partitioniert werden.

Sie können auf die globale Karte zugreifen, indem Sie stateSnapshots() auf dem Rückgabewert mapWithState() aufrufen. Andernfalls wird der Rückgabewert durch die Rückgabewerte Ihrer StateSpecFunction pro Batch bestimmt.

mapWithState() wurde in Spark-1,6, zugegeben, bevor, dass es eine ähnliche Funktion updateStateByKey(), genannt wurde, die meist das gleiche tat, aber durchgeführt auf größere Datensätze schlechter.