Spark-Streaming: Kann nicht Split berechnen, Block nicht

Question

gefunden

Ich versuche, Funken mit Kafka Streaming zu verwenden (Version 1.1.0), aber der Funkenjob hält aufgrund dieser Fehler Absturz:

14/11/21 12:39:23 ERROR TaskSetManager: Task 3967.0:0 failed 4 times; aborting job 
org.apache.spark.SparkException: Job aborted due to stage failure: Task 3967.0:0 failed 4 times, most recent failure: Exception failure in TID 43518 on host ********: java.lang.Exception: Could not compute split, block input-0-1416573258200 not found 
     at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1017) 
     at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1015) 
     at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1015) 
Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3967.0:0 failed 4 times, most recent failure: Exception failure in TID 43518 on host ********: java.lang.Exception: Could not compute split, block input-0-1416573258200 not found 
     at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1017) 
     at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1015) 
     at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1015) 

die einzigen relevanter diese Informationen, die ich aus den Protokollen erhalten ist:

14/11/21 12:34:18 INFO MemoryStore: Block input-0-1416573258200 stored as bytes to memory (size 85.8 KB, free 2.3 GB) 
14/11/21 12:34:18 INFO BlockManagerMaster: Updated info of block input-0-1416573258200 
14/11/21 12:34:18 INFO BlockGenerator: Pushed block input-0-1416573258200 
org.apache.spark.SparkException: Error sending message to BlockManagerMaster [message = GetLocations(input-0-1416573258200)] 
java.lang.Exception: Could not compute split, block input-0-1416573258200 not found 
14/11/21 12:37:35 INFO BlockManagerInfo: Added input-0-1416573258200 in memory on ********:43117 (size: 85.8 KB, free: 2.3 GB) 
org.apache.spark.SparkException: Error sending message to BlockManagerMaster [message = GetLocations(input-0-1416573258200)] 
java.lang.Exception: Could not compute split, block input-0-1416573258200 not found 
org.apache.spark.SparkException: Job aborted due to stage failure: Task 3967.0:0 failed 4 times, most recent failure: Exception failure in TID 43518 on host ********: java.lang.Exception: Could not compute split, block input-0-1416573258200 not found 
java.lang.Exception: Could not compute split, block input-0-1416573258200 not found 
Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3967.0:0 failed 4 times, most recent failure: Exception failure in TID 43518 on host ********: java.lang.Exception: Could not compute split, block input-0-1416573258200 not found 
java.lang.Exception: Could not compute split, block input-0-1416573258200 not found 

Beispielcode:

SparkConf conf = new SparkConf(); 
JavaSparkContext sc = new JavaSparkContext(conf); 
JavaStreamingContext jssc = new JavaStreamingContext(sc, new Duration(5000)); 
jssc.checkpoint(checkpointDir); 

HashMap<String, Integer> topics = new HashMap<String, Integer>(); 
topics.put(KAFKA_TOPIC, 1); 

HashMap<String, String> kafkaParams = new HashMap<String, String>(); 
kafkaParams.put("group.id", "spark-streaming-test"); 
kafkaParams.put("zookeeper.connect", ZOOKEEPER_QUORUM); 
kafkaParams.put("zookeeper.connection.timeout.ms", "1000"); 
kafkaParams.put("auto.offset.reset", "smallest"); 

JavaPairReceiverInputDStream<String, String> kafkaStream = 
    KafkaUtils.createStream(jssc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics, StorageLevels.MEMORY_AND_DISK_SER); 

JavaPairDStream<String, String> streamPair = kafkaStream.flatMapToPair(...).reduceByKey(...); 

Ich bin nicht sicher, welche Ursache dieses Problems ist.

Answer 1

haben Sie versucht mit inputs.persist(StorageLevel.MEMORY_AND_DISK_SER).

z. http://apache-spark-user-list.1001560.n3.nabble.com/Spark-Streaming-Could-not-compute-split-block-not-found-td11186.html

Answer 2

Überprüfen Sie Folgendes.

1) Haben Sie den Streaming-Kontext richtig wie in

def functionToCreateContext(): StreamingContext = { 
    val ssc = new StreamingContext(...) // new context 
    val lines = ssc.socketTextStream(...) // create DStreams 
    ... 
    ssc.checkpoint(checkpointDirectory) // set checkpoint directory 
    ssc 
} 

// Get StreamingContext from checkpoint data or create a new one 
val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _) 

// Do additional setup on context that needs to be done, 
// irrespective of whether it is being started or restarted 
context. ... 

// Start the context 
context.start() 
context.awaitTermination() 

Ihre Initialisierung ist nicht korrekt erstellen.

Werfen Sie einen Blick auf die unten

Beispiel: Code bei recoverableNetworkCount App

2) Haben Sie die Eigenschaft Vorausschreib log "spark.streaming.receiver.writeAheadLog.enable"

3) aktiviert Überprüfen Sie die Stabilität des Streaming in der Streaming-Benutzeroberfläche. Verarbeitungszeit < Batch-Intervall.