Punkty kontrolne przesyłania strumieniowego iskry dla DStreams

Question

W przypadku przesyłania strumieniowego metodą Spark jest możliwe (i obowiązkowe, jeśli zamierzasz używać operacji stanowych), aby ustawić StreamingContext do wykonywania punktów kontrolnych w niezawodne miejsce do przechowywania danych (S3, HDFS, ...) (I):

• Metadane
• DStream rodowód

Jak opisano here, aby ustawić przechowywanie danych wyjściowych trzeba zadzwonić yourSparkStreamingCtx.checkpoint(datastoreURL)

Z drugiej strony można ustawić interwały punktów kontrolnych linii dla każdego DataStream, dzwoniąc pod numer checkpoint(timeInterval). W zasadzie zaleca się, aby ustawić lineage kontrolnego przedziału od 5 do 10 razy przedział przesuwając DataStream „s

dstream.checkpoint (CheckpointInterval). Zazwyczaj interwał kontrolny wynoszący od 5 do 10 przedziałów ślizgowych strumienia DStream jest dobrym ustawieniem na próbę .

Moje pytanie brzmi:

Kiedy kontekst strumieniowych została skonfigurowana do wykonywania punktów kontrolnych i nie ds.checkpoint(interval) nazywa, to linia punktów kontrolnych włączona dla wszystkich strumieni danych z domyślnym checkpointInterval równej batchInterval? Czy przeciwnie, tylko sprawdzanie punktów metadanych, co jest włączone?

Answer 1

Sprawdzanie kodu Spark (v1.5) stwierdziliśmy, że DStream s'checkpoint są włączone w dwóch przypadkach:

Przez wyraźnego wezwania do ich sposobu checkpoint (nie StreamContext „s):

/** 
* Enable periodic checkpointing of RDDs of this DStream 
* @param interval Time interval after which generated RDD will be checkpointed 
*/ 
def checkpoint(interval: Duration): DStream[T] = { 
    if (isInitialized) { 
     throw new UnsupportedOperationException(
      "Cannot change checkpoint interval of an DStream after streaming context has started") 
    } 
    persist() 
    checkpointDuration = interval 
    this 
} 

na inicjalizacji DStream tak długo jak podklasa betonowego „DStream” ma przesłonięte mustCheckpoint atrybut (ustawienie go na true):

private[streaming] def initialize(time: Time) { 
    ... 
    ... 
    // Set the checkpoint interval to be slideDuration or 10 seconds, which ever is larger 
    if (mustCheckpoint && checkpointDuration == null) { 
    checkpointDuration = slideDuration * math.ceil(Seconds(10)/slideDuration).toInt 
    logInfo("Checkpoint interval automatically set to " + checkpointDuration) 
    } 
    ... 

Pierwszy przypadek jest oczywisty. Wykonywanie analiz na naiwne Spark kodu Streaming:

grep "val mustCheckpoint = true" $(find -type f -name "*.scala") 

> ./org/apache/spark/streaming/api/python/PythonDStream.scala: override  val mustCheckpoint = true 
>./org/apache/spark/streaming/dstream/ReducedWindowedDStream.scala: override val mustCheckpoint = true 
>./org/apache/spark/streaming/dstream/StateDStream.scala: override val mustCheckpoint = true 

mogę znaleźć, że w ogóle (pomijając PythonDStream) StreamingContext kontrolny umożliwia tylko punkty kontrolne Lineage dla StateDStream i ReducedWindowedDStream przypadkach. Te przypadki są rezultatem przemian (odpowiednio, a):

• updateStateByKey: Oznacza to, że strumień zapewniając stan przez kilka okien.
• reduceByKeyAndWindow