flink如flink如何从流中取出自定义的数据结构出问题

flink版本使用1.12.2。有一个需求就是想要从stream中拿出自定义的数据结构，暂且叫a并赋值给后面变量，基于这个a取他的属性作一些判断操作。 比如： val ds: DataStream[b] = stream.filter(_.nonEmpty).map(new MapFunction[String, b] {

override def map(value: String) = { val recallKafka = JSON.parseObject(value, classOf[a])

b(recallKafka.group_id, value, recallKafka.eventTime)

} })

val kafkaCommonData: a =recallKafka 判断条件 if (kafkaCommonData.data.date != null) {xxxxx} if (kafkaCommonData.data.userinfo != null) {xxxx} ..... 请问一下，我通过什么方法能单独把流中的某个数据结构给取出来呢？如果有方式的话应该要怎么写呢？大佬们帮忙看一下啊，卡了好几天 了，难受。。*来自志愿者整理的flink邮件归档