// stream contains stuff like these in a flink tuple
//(custom deserializer of array to tuple2???)
val jsonExample = """["foo", "bar"]"""
val stream: DataStream[Tuple2[JString, JString]] = ???
val filteredStream = stream.filter(x => x.getField(0).extract[String] == "foo")
2条答案
按热度按时间olhwl3o21#
这个问题对我来说似乎有点太不明确了,但也许,这不管用吗?
如果你在写scala,最好不要使用flink元组。选择case类或者至少scala元组?
ergxz8rk2#
不用Map到元组,只需Map到case类并过滤掉不需要的内容: