Kıvılcımda Yakalanmamış İstisna İşleme

Question

Kafka konusuna giren mesajlara cevap veren Java tabanlı bir Spark Streaming uygulaması üzerinde çalışıyorum. Her mesaj için, uygulama bazı işlemleri yapar ve sonuçları farklı bir Kafka konusuna yazar.

Bazen beklenmedik verilerle ilgili sorunlar nedeniyle, RDD'lerde çalışan kod başarısız olabilir ve bir istisna atar. Bu olduğunda, gerekli işlemi yapabilecek ve bir hata konusuna bir mesaj bırakabilecek genel bir işleyiciye sahip olmak isterim. Şu anda, bu istisnalar Spark’in Spark’in log’unda yazılıyor.

Bunu yapmak için en iyi yaklaşım nedir, RDD'lerde çalışan her kod bloğu için try-catch blokları yazmak yerine?

Answer 1

Bunu yapan genel bir işlev yazabilirsiniz. Sadece, sadece Spark istisnalarını atabilecek olanlar olan RDD eylemleri etrafında sarmanız gerekir (.map ve .filter gibi transformatörler, eylemler tarafından tembellik ile yürütülür).

(Bunun Scala'da olduğunu varsayarsak) Hatta bir şeyleri örtülü olarak da deneyebilirsiniz. RDD'yi tutan ve hatayı işleyen bir sınıf oluşturun. İşte böyle görünüşüne ait bir kroki: Sen failsafeAction içine girmesine ya da hata konu mesajlaşma ekleyebilir

implicit class FailSafeRDD[T](rdd: RDD[T]) { 
    def failsafeAction[U](fn: RDD[T] => U): Try[U] = Try { 
    fn(rdd) 
    } 
} 

Eğer başarısızlık her zaman yapmak istiyorum. gerçi Bunun yanı sıra

val rdd = ??? // Some rdd you already have 
val resultOrException = rdd.failsafeAction { r => r.count() } 

, ben "en iyi" yaklaşımı uygulama ihtiyaçlarına biraz sübjektif hayal: Ve sonra kullanım gibi olabilir.

Answer 2

I =>

dstream.foreachRDD { case rdd: RDD[String] => 
    rdd.foreach { case string: String => 
     try { 
     val kafkaProducer = ... 
     val msg = ... 
     kafkaProducer.send(msg) 
     } catch { 
     case d: DataException=> 
      val kafkaErrorProducer = ... 
      val errorMsg = ... 
      kafkaErrorProducer.send(errorMsg) 
     case t: Throwable => 
      //further error handling 
     } 
    } 
}