Как опустить исключение нулевого значения в flink-kafka, любая помощь будет делать - PullRequest
Новая
       39

Как опустить исключение нулевого значения в flink-kafka, любая помощь будет делать

1 голос
/ 04 октября 2019

Я пытаюсь создать код, который создает предупреждение, когда температура превышает пороговую температуру (как определено в коде), но ключевой поток создает проблему. Я новичок, чтобы моргать и промежуточный в скале. Мне нужна помощь в этом коде

Я перепробовал почти все 

 def main(args: Array[String]): Unit = {
    val TEMPERATURE_THRESHOLD: Double = 50.00

    val see: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("zookeeper.connect", "localhost:2181")

    val src = see.addSource(new FlinkKafkaConsumer010("broadcast",
      new JSONKeyValueDeserializationSchema(true), properties))


    var data = src.map { v => {
      val loc = v.get("locationID").asInstanceOf[String]
      val temperature = v.get("temp").asDouble()
    //val json = Map("locationID" -> locationID, "temp" -> v.temp)
    //val jsonVal = JSONObject(json).toString()
      (loc, temperature)
    }}
 data = data
      .keyBy(
        v => v._1
      )

{"locationID": "ASK", "temp": 35} {"locationID": "BC", "temp": 45} {"locationID": "CHD", "temp": 55} {"locationID": "RAJ", "temp": 65} {"locationID": "EGY", "temp": 55}

Я думаю, мой кортеж где-то принимает нулевое значение, мне нужна помощь, чтобы опустить эту ошибку

Ошибка: - 

Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
...

    at flinkBroadcast1$.main(flinkBroadcast1.scala:59)
    at flinkBroadcast1.main(flinkBroadcast1.scala)
Caused by: java.lang.Exception: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
...
Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
...
Caused by: java.lang.NullPointerException
...