स्पार्क

Question

में अनकॉच अपवाद हैंडलिंग मैं जावा आधारित स्पार्क स्ट्रीमिंग एप्लिकेशन पर काम कर रहा हूं जो एक कफका विषय के माध्यम से आने वाले संदेशों का जवाब देता है। प्रत्येक संदेश के लिए, एप्लिकेशन कुछ प्रसंस्करण करता है, और परिणामों को एक अलग काफ्का विषय पर वापस लिखता है।

कभी-कभी अप्रत्याशित डेटा से संबंधित मुद्दों के कारण, आरडीडी पर चलने वाला कोड असफल हो सकता है और अपवाद फेंक सकता है। जब ऐसा होता है, तो मैं एक सामान्य हैंडलर रखना चाहता हूं जो आवश्यक कार्रवाई कर सके और एक त्रुटि विषय को संदेश छोड़ दे। अभी, इन अपवादों को स्पार्क के लॉग में स्पार्क द्वारा स्वयं लिखा गया है।

आरडीडी पर काम करने वाले प्रत्येक कोड ब्लॉक के लिए प्रयास-पकड़ ब्लॉक लिखने के बजाय ऐसा करने का सबसे अच्छा तरीका क्या है?

Answer 1

आप एक सामान्य कार्य लिख सकते हैं जो यह करता है। आपको केवल इसे आरडीडी कार्यों के आसपास लपेटने की ज़रूरत है क्योंकि वे केवल एकमात्र हैं जो स्पार्क अपवादों को फेंक सकते हैं (ट्रांसफॉर्मर जैसे .map और .filter क्रियाओं द्वारा आलसी निष्पादित हैं)।

(मान लीजिए कि यह स्कैला में है) आप शायद इम्प्लिकेट्स के साथ कुछ भी कोशिश कर सकते हैं। एक कक्षा बनाएं जो आरडीडी रखती है और त्रुटि को संभालती है। यहां उसका कैसा लग सकता है की एक संक्षिप्त वर्णन:

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

आप failsafeAction या कुछ में त्रुटि विषय संदेश सेवा जोड़ सकता है आप विफलता पर हर बार करना चाहते हैं। और फिर उपयोग इस तरह हो सकता है:

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

इसके अलावा, मुझे लगता है कि "सर्वश्रेष्ठ" दृष्टिकोण कुछ हद तक आवेदन आवश्यकताओं के लिए व्यक्तिपरक है।

Answer 2

मुझे लगता है कि आप भी एक कोशिश पकड़ के साथ इस को लागू कर सकता है =>

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 
     } 
    } 
}