Sparkでキャッチされない例外処理

Question

私は、カフカのトピックから来るメッセージに応答するJavaベースのSpark Streamingアプリケーションを開発中です。各メッセージについて、アプリケーションは何らかの処理を行い、その結果を別のカフカのトピックに書き戻します。

予期しないデータ関連の問題により、RDDで動作するコードが失敗し、例外がスローされることがあります。それが起こると、必要なアクションを実行してエラーのトピックにメッセージをドロップするジェネリックハンドラが必要になります。今のところ、これらの例外はスパーク自体によってスパークのログに書き込まれます。

RDDで動作するコードブロックごとにtry-catchブロックを書き込むのではなく、これを実行する最良の方法は何ですか？

Answer 1

これを行う汎用関数を書くことができます。スパーク例外を発生させる唯一のもの（.mapや.filterなどのトランスフォーマーはアクションによって遅延実行される）がRDDアクションの周りにラップする必要があります。

（これはScalaにあると仮定します）おそらく暗黙のうちに何かを試すこともできます。 RDDを保持し、エラーを処理するクラスを作成します。ここではそれがどのように見えるかのスケッチ：あなたはfailsafeActionか何かにエラートピックメッセージを追加することができます

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

あなたが失敗した場合に毎回やりたいです。使用方法は次のようになります：

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