Stwórz stanowy łańcuch dla różnych zdarzeń i przypisz globalny identyfikator w iskrze

Question

Pracujemy z spark 1.6 i staramy się zachować globalną tożsamość dla podobnych zdarzeń. Może istnieć kilka "grup" zdarzeń o identycznym ID (w przykładzie jako liczba, litery są dodawane tylko dla unikalności). I wiemy, że niektóre z tych wydarzeń są podobne, więc jesteśmy w stanie je połączyć. Chcemy zachować coś takiego:

Z -> 1, 2, 3 
X -> 4 

więc w przyszłości, jeżeli pewne zdarzenia z identyfikatorem 4 przyjdą możemy przypisać X jako globalnej tożsamości.

Proszę sprawdzić przykład dla lepszego zobrazowania:

Powiedzmy mamy jakieś dane strumieniowe przyjście do pracy zapłonowej.

Ponieważ wydarzenie 1 to nasz pierwszy występ, chcemy przypisać 1 to Z. Następnie wiemy, że 1b i 2c są podobne. więc chcemy zachować mapowanie. To samo dotyczy 2e i 3f, więc potrzebujemy mapowania 3-2. Na razie mamy 3 pary: 1->Z, 2->1, .

I chcemy stworzyć "historyczną" ścieżkę: Z <- 1 <- 2 <- 3 Na koniec będziemy mieli wszystkie zdarzenia z ID = Z.

1a -> Z 
1b -> Z 
2c -> Z 
2d -> Z 
2e -> Z 
3f -> Z 
3g -> Z 
3h -> Z 
4i -> X 

Staraliśmy się wykorzystać mapwithstate ale jedyną rzeczą, byliśmy w stanie to zrobić, że 2->1 i 3->2. Z mapwithstate nie byliśmy w stanie uzyskać stanu dla "rodzica" w stanie dla bieżącego zdarzenia - np. bieżące zdarzenie 3 z rodzicem 2 i nie może uzyskać 2 -> 1 ani żadnego 1 -> Z.

Czy możliwe jest wykonanie globalnego mapowania? Wypróbowaliśmy już akumulatory i emisję, ale wygląda na to, że nie nadaje się zbyt dobrze. I nie byliśmy w stanie zastąpić zdarzeń 1 dla pierwszego mapowania i zdarzeń 2 dla drugiego mapowania z Z.

Jeśli pojawi się nowe zdarzenie 5 i będzie podobne z 3h, na przykład musimy ponownie przypisać mapowanie 5->Z.

Answer 1

Poniżej znajduje się rozwiązanie dla danego problemu, używając zmiennego odniesienia do "stanu" RDD, który za każdym razem aktualizujemy za pomocą nowych wyników.

Używamy transform do oznaczania przychodzącego strumienia zdarzeń z unikalnym identyfikatorem globalnym, wykonując sprzężenie podobieństwa. Jest to połączenie "ręcznie", w którym używamy produktu z dwóch zestawów danych i porównujemy każdą pozycję parami.

Należy pamiętać, że jest to kosztowny proces. Istnieje wiele części, które można zmienić, w zależności od specyfiki oczekiwanego strumienia. Na przykład możemy zastąpić globalny stan RDD przez lokalny map i zastosować map-side do łączenia szybszego podobieństwa, ale to bardzo zależy od oczekiwanej liczności zbioru unikalnych identyfikatorów.

To było trudniejsze, niż się początkowo spodziewałem. Traktuj to tylko jako punkt wyjścia do bardziej niezawodnego rozwiązania. Na przykład operacja union na stanie RDD wymaga regularnego sprawdzania punktów, aby uniknąć wzrostu kontroli DAG. (Jest dużo miejsca na ulepszenia - ale to jest ponad rozsądny wysiłek, aby udzielić odpowiedzi.)

Tu naszkicować rdzeń rozwiązania dla całego badanego notebooka zobaczyć UniqueGlobalStateChains.snb

// this mutable reference points to the `states` that we keep across interations  
@transient var states: RDD[(String, (Int, Long))] = sparkContext.emptyRDD 

// we assume an incoming Event stream. Here we prepare it for the global id-process 
@transient val eventsById = eventStream.map(event => (event.id, event)) 
@transient val groupedEvents = eventsById.groupByKey() 

// this is the core of the solution. 
// We transform the incoming events into tagged events. 
// As a by-product, the mutable `states` reference will get updated with the latest state mapping. 
// the "chain" of events can be reconstructed ordering the states by timestamp 

@transient val taggedEvents = groupedEvents.transform{ (events, currentTime) => 
    val currentTransitions = states.reduceByKey{case (event1, event2) => Seq(event1, event2).maxBy{case (id, ts) => ts}}       
    val currentMappings = currentTransitions.map{case (globalId, (currentId, maxTx)) => (currentId, globalId)} 

    val newEventIds = events.keys // let's extract the ids of the incoming (grouped) events 
    val similarityJoinMap = newEventIds.cartesian(currentMappings) 
     .collect{case (eventId, (currentId, globalId)) if (isSimilar(currentId)(eventId)) => (eventId, globalId)} 
     .collectAsMap 
    //val similarityBC = sparkContext.broadcast(similarityJoinMap)     
    val newGlobalKeys = newEventIds.map(id => (id, similarityJoinMap.getOrElse(id, genGlobalId()))) 
    newGlobalKeys.cache() //avoid lazy evaluation to generate multiple global ids 

    val newTaggedEvents = events.join(newGlobalKeys).flatMap{case (eventId, (events, globalKey)) => 
             events.map(event => (event.id,event.payload, globalKey)) 
            } 
    val newStates = newGlobalKeys.map{case (eventId, globalKey) => (globalKey, (eventId, currentTime.milliseconds))} 
    currentState = newStates    
    states.unpersist(false)        
    states = newStates.union(states) 
    states.cache()        
    newTaggedEvents 
    } 

Biorąc pod uwagę to sekwencja wejściowa:

"1|a,1|b,3|c", "2|d,2|e,2|f", "3|g,3|h,3|i,4|j", "5|k", "4|f,1|g", "6|h" 

Dostajemy:

oznaczone wydarzenia z globalny identyfikator:

--- 
1|a: gen-4180,1|b: gen-4180,3|c: gen-5819 
--- 
2|d: gen-4180,2|e: gen-4180,2|f: gen-4180 
--- 
3|g: gen-4180,3|h: gen-4180,3|i: gen-4180,4|j: gen-5819 
--- 
5|k: gen-5819 
--- 
1|g: gen-2635,4|f: gen-4180 
--- 
6|h: gen-5819 

I możemy zrekonstruować łańcuch zdarzeń, które pochodzą z globalnej ID:

gen-4180: 1<-2<-3<-4 
gen-2635: 1 
gen-5819: 3<-4<-5<-6 

-o-