Pause Spark Streaming Job

Question

Ich habe einfache Spark-Streaming-Anwendung, die Daten aus Kafka liest und senden Sie diese Daten nach der Transformation auf einem HTTP-Endpunkt (oder ein anderes Kafka - für diese Frage lassen Sie uns http betrachten). Ich überweise Aufträge mit job-server.

Ich beginne gerade den Verbrauch von Quelle kafka mit "auto.offset.reset" = "kleinste" und Intervall = 3s. Im glücklichen Fall sieht alles gut aus. Hier ein Auszug:

kafkaInputDStream.foreachRDD(rdd => { 
    rdd.foreach(item => { 
    //This will throw exception if http endpoint isn't reachable 
     httpProcessor.process(item._1, item._2) 
    }) 
}) 

Da "auto.offset.reset" = "kleinste", diese Prozesse zu 200K Nachrichten in einem Job. Wenn ich den HTTP-Server-Mid-Job stoppe (ein Problem in POSTing simuliert) und httpProcessor.process die Ausnahme auslöst, schlägt dieser Job fehl und alles, was nicht verarbeitet wird, ist verloren. Ich sehe, dass es danach alle 3 Sekunden pollt.

Also meine Frage ist: Wenn in der nächsten 3 Sekunden Job

1. meine Vermutung ist richtig, dass, wenn es X-Nachrichten bekommt und nur Y, bevor man einen Fehler verarbeitet werden kann, Rest X-Y wird nicht bearbeitet?
2. Gibt es eine Möglichkeit, den Stream/Verbrauch von Kafka anzuhalten? Zum Beispiel, wenn es ein intermittierendes Netzwerkproblem gibt und höchstwahrscheinlich alle verbrauchten Nachrichten in dieser Zeit verloren gehen. Etwas, das immer wieder versucht (möglicherweise exponentielles Zurücksetzen) und wann auch immer der http-Endpunkt UP ist, fange wieder an zu konsumieren.

Dank

Answer 1

denke ich, Frühling Wolke Strom Ihr Problem lösen könnte. Kafka ist Quelle. Spark Streaming ist ein Prozessor. Http ist sinken. Nur wenn eine Eingabe von Kafka vorliegt, wird Spark Streaming verarbeitet. Sie müssen die Eingabe von Kafka nicht stoppen oder fortsetzen. Hoffe es hilft.

Answer 2

Ja, Ihre Annahme ist richtig, dass, wenn Ihre Partition ausfällt, die verbleibenden Ereignisse für den Moment nicht verarbeitet werden.

Es gibt jedoch einige Parameter, die Sie einstellen müssen, um das gewünschte Verhalten zu erzielen (wenn Sie DirectKafkaInputDStream verwenden).

Beginnen wir mit "auto.offset.reset" = "kleinste": Dieser Parameter sagt Kafka, von Anfang an zu beginnen, wenn es kein gespeichertes Commit für die aktuelle Gruppe gibt. Da Sie erwähnt haben, dass Ihre RDD nach dem Start eine Menge Nachrichten enthält, gehe ich davon aus, dass Sie Ihre Nachrichten nicht korrekt übertragen. Wenn Sie erwarten, genau einmal Semantik, sollten Sie auf jeden Fall in Betracht ziehen, Ihre Offsets zu verfolgen, da der DirectKafkaStreamInput dies explizit nicht verfolgt.

Startverschiebungen im Voraus festgelegt werden, und diese DSTREAM ist für die Begehung von Versetzungen nicht verantwortlich, so dass Sie genau einmal

Comment in the DirectKafkaInputSream Branch 1.6

, dass Ihre Nachricht erneut verarbeitet werden derzeit die steuern können, Bei jedem Neustart des Streaming-Jobs.

Wenn Sie Ihre verarbeiteten Offsets festschreiben und beim Start in den InputDStream übergeben, wird der Listener mit dem letzten festgeschriebenen Offset fortgesetzt.

In Bezug auf Rückstau, die DirectKafkaInputDStream nutzt bereits ein RateController, die schätzt, wie viele Ereignisse in einer Charge verarbeitet werden sollen.

es nutzen zu können, müssen Sie Rückstau aktivieren:

"spark.streaming.backpressure.enabled": true 

Sie können auch die „spark.streaming.kafka.maxRatePerPartition“ begrenzen eine Obergrenze für die Chargengröße hinzuzufügen.

Wenn Sie den Gegendruck selbst regeln möchten (und vielleicht den Verbraucher für eine Weile komplett stoppen), sollten Sie einige Methoden von StreamingListener implementieren und in Ihrer Arbeit verwenden. Sie können z.B. Entscheiden Sie sich nach jedem abgeschlossenen Batch, um den Streaming-Job zu stoppen oder nicht mit dem StreamingListener.