在本篇文章中我们接着来说filter函数,此函数主要作用就是根据用户条件，过滤数据流中数据。
需求
将数据流的单词做大写转换，并保留其小写，然后只打印出字符串长度大于5的元素。
解决方案
代码语言：javascript复制public class StreamTest {
private static final Logger LOG = LoggerFactory.getLogger(StreamTest.class);
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
String[] words = {"apple","orange","banana","watermelon"};
//创建DataSource
//根据给定的元素创建一个数据流，元素必须是相同的类型，比如全部为String，或者全部为int。
DataStreamSource<String> ds = env.fromElements(words);
//Transformations
//对DataStream应用一个flatMap转换。对DataStream中的每一个元素都会调用FlatMapFunction接口的具体实现类。flatMap方法可以返回任意个元素，当然也可以什么都不返回。
SingleOutputStreamOperator<String> flatMap = ds.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out) throws Exception {
out.collect(value);
out.collect(value.toUpperCase());
}
});
//过滤掉单词长度不大于5的单词
//对数据流中的每个元素执行filter方法，只通过结果为True的元素
DataStream<String> filter = flatMap.filter((value) -> value.length()>5);
//sinks打印出信息
//给DataStream添加一个Sinks
filter.addSink(new SinkFunction<String>() {
@Override
public void invoke(String value) throws Exception {
LOG.info(value);
}
});
env.execute("Flink Streaming Java API Skeleton");
}
}执行效果
代码语言：javascript复制13:59:46,968 INFO fun.cosmozhu.session4.StreamTest - banana
13:59:46,969 INFO fun.cosmozhu.session4.StreamTest - BANANA
13:59:46,968 INFO fun.cosmozhu.session4.StreamTest - orange
13:59:46,969 INFO fun.cosmozhu.session4.StreamTest - ORANGE
13:59:46,968 INFO fun.cosmozhu.session4.StreamTest - watermelon
13:59:46,969 INFO fun.cosmozhu.session4.StreamTest - WATERMELON代码地址
代码语言：javascript复制https://github.com/chaoxxx/learn-flink-stream-api/blob/master/src/main/java/fun/cosmozhu/session4/StreamTest.java作者：cosmozhu --90后的老父亲，专注于保护地球的程序员
个人网站：https://www.cosmozhu.fun
欢迎转载，转载时请注明出处。
相关文章
- Flink-1.9流计算开发：十六、intervalJoin函数
- Flink-1.9流计算开发：十五、join函数
- Flink-1.9流计算开发：十四、union函数
- Flink-1.9流计算开发：十三、min、minBy、max、maxBy函数
- Flink-1.9流计算开发：十二、apply函数
