我使用datasetapi，我有两种case类
case class Geo(country:Int, province:Int, city:Int, county:Int)
case class AntiFraudLog(
eventType: Int,
valid: Boolean
)
case class AntiFraudSession(fraudLogs: Seq[AntiFraudLog])然后我生成了一个键/值对，它的值是case类。
val dataKeyValue: DataSet[(Long, AntiFraudLog)]并尝试对具有相同键的项进行分组
val groupedSortedData = dataKeyValue groupBy 0然后将分组数据转换为另一个case类
val sessionData:DataSet[AntiFraudSession] = groupedSortedData reduceGroup(
logs => AntiFraudSession(logs.map(_._2).toSeq)
)但是当我运行程序时，我遇到了这样的异常
Caused by: java.lang.NullPointerException
at org.apache.flink.api.scala.typeutils.TraversableSerializer.serialize(TraversableSerializer.scala:90)
at org.apache.flink.api.scala.typeutils.TraversableSerializer.serialize(TraversableSerializer.scala:32)
at org.apache.flink.api.scala.typeutils.CaseClassSerializer.serialize(CaseClassSerializer.scala:100)
at org.apache.flink.api.scala.typeutils.CaseClassSerializer.serialize(CaseClassSerializer.scala:30)
at org.apache.flink.api.scala.typeutils.CaseClassSerializer.serialize(CaseClassSerializer.scala:100)
at org.apache.flink.api.scala.typeutils.CaseClassSerializer.serialize(CaseClassSerializer.scala:30)
at org.apache.flink.runtime.plugable.SerializationDelegate.write(SerializationDelegate.java:56)
at org.apache.flink.runtime.io.network.api.serialization.SpanningRecordSerializer.addRecord(SpanningRecordSerializer.java:83)
at org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:85)
at org.apache.flink.runtime.operators.shipping.OutputCollector.collect(OutputCollector.java:65)
at org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
at org.apache.flink.api.scala.DataSet$$anon$5$$anonfun$flatMap$1.apply(DataSet.scala:417)
at org.apache.flink.api.scala.DataSet$$anon$5$$anonfun$flatMap$1.apply(DataSet.scala:417)
at scala.collection.immutable.List.foreach(List.scala:381)
at org.apache.flink.api.scala.DataSet$$anon$5.flatMap(DataSet.scala:417)
at org.apache.flink.runtime.operators.chaining.ChainedFlatMapDriver.collect(ChainedFlatMapDriver.java:80)
at org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:163)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
at java.lang.Thread.run(Thread.java:745)有人知道怎么修吗？
2条答案
按热度按时间mqkwyuun1#
flink似乎无法序列化具有空值集合字段的case类。在您的场景中，这将是一个fraudlogs=null的会话。您是否认为有更多的转换逻辑会导致sessiondata中出现类似的元素？
xesrikrc2#
请确保中没有空对象/值
case-class在flink中使用scala反序列化时。避免
java.lang.NullPointerException使用Option在这种情况下，类字段/对象将为null。从你的例子来看：
如果任何字段为空
如果case类对象为空
注意：在scala中使用null不是一个好的实践/标准。因此，请尝试使用scala中提供的许多其他选项来避免这种情况。
有关更多详细信息，请单击此处。