Handle verschiedene Zustände

Question

Ich fragte mich, ob es möglich war, verschiedene Zustände in einer Anwendung radikal zu erhalten? Zum Beispiel, haben die update function des ersten Zustandes den einen aus dem zweiten Zustand?

Ich erinnere mich nicht durch ein solches Beispiel gehen, noch fand ich eine Gegenanzeige ... Basierend auf dem Beispiel von https://docs.cloud.databricks.com/docs/spark/1.6/examples/Streaming%20mapWithState.html, weiß ich keinen Grund, warum ich nicht anders trackStateFunc s mit anderen haben könnte State s, und aktualisieren Sie immer noch diese dank ihrer Key, wie unten dargestellt:

def firstTrackStateFunc(batchTime: Time, 
         key: String, 
         value: Option[Int], 
         state: State[Long]): Option[(String, Long)] = { 
    val sum = value.getOrElse(0).toLong + state.getOption.getOrElse(0L) 
    val output = (key, sum) 
    state.update(sum) 
    Some(output) 
} 

und

def secondTrackStateFunc(batchTime: Time, 
         key: String, 
         value: Option[Int], 
         state: State[Int]): Option[(String, Long)] = { 
    // disregard problems this example would cause 
    val dif = value.getOrElse(0) - state.getOption.getOrElse(0L) 
    val output = (key, dif) 
    state.update(dif) 
    Some(output) 
} 

ich denke, dass dies möglich ist, aber immer noch unsicher. Ich möchte jemanden bestätigen oder diese Annahme ungültig ...

Answer 1

ich mich gefragt, ob es möglich war, über eine Anwendung radikal andere Staaten aufrecht zu erhalten?

Jeder Aufruf zu mapWithState auf einem DStream[(Key, Value)] kann man State[T] Objekt halten. Diese T muss für jeden Aufruf von mapWithState gleich sein. Um verschiedene Zustände zu verwenden, können Sie entweder mapWithState Anrufe ketten, wo man Option[U] anothers Eingang ist, oder Sie können die DStream teilen und einen unterschiedlichen mapWithState Anruf auf jeden anwenden. Sie können jedoch kein anderes State[T] Objekt in einem anderen Objekt aufrufen, da diese voneinander isoliert sind und der Zustand des anderen nicht mutiert werden kann.

Answer 2

@Yuval gab eine gute Antwort auf Chain MapWithState Funktionen. Ich habe jedoch einen anderen Ansatz. Anstelle von zwei mapWithState-Aufrufen können Sie sowohl die Summe als auch das Diff in denselben Status [(Int, Int)] setzen.

In diesem Fall würden Sie nur eine mapWithState Funktionen benötigen, wo Sie beide Dinge aktualisieren können. Etwas wie dieses:

def trackStateFunc(batchTime: Time, 
        key: String, 
        value: Option[Int], 
        state: State[(Long, Int)]): Option[(String, (Long, Int))] = 
{ 
    val sum = value.getOrElse(0).toLong + state.getOption.getOrElse(0L) 
    val dif = value.getOrElse(0) - state.getOption.getOrElse(0L) 
    val output = (key, (sum, diff)) 
    state.update((sum, diff)) 
    Some(output) 
}