RuntimeException: не удалось извлечь ключ происходит только в среде выполнения - PullRequest
Новая
       12

RuntimeException: не удалось извлечь ключ происходит только в среде выполнения

0 голосов
/ 20 ноября 2018

Я использую Flink локально на моей машине, я получаю исключение ниже при чтении из темы Кафки.при запуске из ide (intellij) он работает отлично.однако, когда я развертываю свой jar-файл для flink среды выполнения (локально), используя 

/bin/flink run ~MyApp-1.0-SNAPSHOT.jar

, мой класс выглядит следующим образом 

case class Foo(id: String, value: String, timestamp: Long, counter: Int)

Я получаю это исключение 

java.lang.RuntimeException: Could not extract key from Foo(some-uuid,some-text,1540348398,1)
    at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:110)
    at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collect(RecordWriterOutput.java:89)
    at org.apache.flink.streaming.runtime.io.RecordWriterOutput.collect(RecordWriterOutput.java:45)
    at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:689)
    at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:667)
    at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:40)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:579)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
    at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:689)
    at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:667)
    at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
    at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111)
    at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordWithTimestamp(AbstractFetcher.java:398)
    at org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher.emitRecord(Kafka010Fetcher.java:89)
    at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:154)
    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:738)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:87)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:56)
    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: Could not extract key from Foo("some-uuid","text",1540348398,1)
    at org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.selectChannels(KeyGroupStreamPartitioner.java:61)
    at org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.selectChannels(KeyGroupStreamPartitioner.java:32)
    at org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:106)
    at org.apache.flink.streaming.runtime.io.StreamRecordWriter.emit(StreamRecordWriter.java:81)
    at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:107)
    ... 22 more
Caused by: java.lang.NullPointerException
    at com.bluevoyant.StreamingJob$$anonfun$3.apply(StreamingJob.scala:41)
    at com.bluevoyant.StreamingJob$$anonfun$3.apply(StreamingJob.scala:40)
    at org.apache.flink.streaming.api.scala.DataStream$$anon$2.getKey(DataStream.scala:411)
    at org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.selectChannels(KeyGroupStreamPartitioner.java:59)
    ... 26 more

мой ключевой раздел прост (partitionFactor = некоторое число) 

env.addSource(kafkaConsumer)
    .filter(_.id != null)
    .keyBy{ r =>
        val h = fastHash(r.id) % partitionFactor
        math.abs(h)
    }
    .map(...)

Опять же, это происходит только во время выполнения, а не когда я запускаю его из-за intellij это так расстраивает, любой совет?

...