FlinkKafkaConsumer - Scala использует JSON-сообщение - PullRequest
Новая
       31

FlinkKafkaConsumer - Scala использует JSON-сообщение

0 голосов
/ 12 февраля 2019

Я новичок в скале и моргании.Я использую следующий код, чтобы использовать сообщение json из потока kafka.

object Job {
  def main(args: Array[String]) {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    // only required for Kafka 0.8
    properties.setProperty("zookeeper.connect", "localhost:2181")
    val topic: String = "test"

    val consumer = new FlinkKafkaConsumer[ObjectNode](topic, new JSONKeyValueDeserializationSchema(false), properties)
    // set up the execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream = env.addSource(consumer).name("opendental")
      .enableCheckpointing(5000) // checkpoint every 5000 msecs
      .print()

    // execute program
    env.execute("Flink Scala API Skeleton")
  }
}

Я использую команду mvn clean package для компиляции кода Scala в .jar, но сборка завершается неудачно из-за перегруженного метода.Это ошибка 

[ERROR]   (x$1: java.util.regex.Pattern,x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: java.util.regex.Pattern,x$2: (some other)org.apache.flink.api.common.serialization.DeserializationSchema(in <none>)[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: java.util.List[String],x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: java.util.List[String],x$2: org.apache.flink.api.common.serialization.DeserializationSchema(in <none>)[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: String,x$2: org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode] <and>
[ERROR]   (x$1: String,x$2: org.apache.flink.api.common.serialization.DeserializationSchema[com.fasterxml.jackson.databind.node.ObjectNode],x$3: java.util.Properties)org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer[com.fasterxml.jackson.databind.node.ObjectNode]
[ERROR]  cannot be applied to (String, org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema, java.util.Properties)
[ERROR]     val consumer = new FlinkKafkaConsumer[ObjectNode](topic, new JSONKeyValueDeserializationSchema(false), properties)
[ERROR]

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

...