Flink java.lang.ClassNotFoundException: play.api.libs.json.Reads

Question

Я пишу код, который читает из текстового файла и анализирует каждую строку как Json, но задание в Flink (1.0.3 - Cluster 2 Taskmanagers) терпит неудачу с го excetion: java.lang.ClassNotFoundException: play.api.libs.json.Reads

Мой код:

import org.apache.flink.streaming.api.scala._ 
import play.api.libs.json.Json 
import org.joda.time.{DateTime, DateTimeZone} 

object rabbitjob { 

case class MyJson(pr: Long, 
       dv: Long, 
       ty: Int, 
       cr: String, 
       rc: String, 
       vl: Boolean, 
       ss: String, 
       id: Long 
       ) 

def main (args:Array[String]){ 
    val env = StreamExecutionEnvironment.getExecutionEnvironment 
    val stream = env.readTextFile("hdfs:///test/ignicion.io") 

    implicit val myJsonReads = Json.reads[MyJson] 

    def jsontr(cuerpo: String): Unit ={ 

    val inputJson = Json.parse(cuerpo) 
    val myJsonInstance: MyJson = inputJson.as[MyJson] 

    println(DateTime.now(DateTimeZone.UTC).getMillis() + " " + myJsonInstance.cr + " " + matchtype(myJsonInstance.ty) + " " + " " + matchvalue(myJsonInstance.vl)) 

    def matchtype (x: Int): String = x match{ 
     case 1 => "Door" 
     case 2 => "Window" 
     case _ => "otros" 
    } 

    def matchvalue (x: Boolean): String = x match{ 
     case true => "ON" 
     case false => "OFF" 
    } 
    } 

    println(stream) 
    stream.map(jsontr(_)) 

env.execute("Test Rabbit") 
} 
} 

Любая идея об этом ?? Заранее спасибо

Answer 1

Вам необходимо упаковать все внешние зависимости вместе с вашей программой Flink в «толстый» JAR-файл, который вы отправляете в кластер.

В документации Flink показано, как это сделать с помощью Maven.