首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Flink-1.9流计算开发:四、filter函数

Flink-1.9流计算开发:四、filter函数

作者头像
cosmozhu
发布2020-06-15 06:32:32
1.4K0
发布2020-06-15 06:32:32
举报
文章被收录于专栏:cosmozhu技术篇cosmozhu技术篇

在本篇文章中我们接着来说filter函数,此函数主要作用就是根据用户条件,过滤数据流中数据。

需求

将数据流的单词做大写转换,并保留其小写,然后只打印出字符串长度大于5的元素。

解决方案

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");
    }
}

执行效果

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

代码地址

https://github.com/chaoxxx/learn-flink-stream-api/blob/master/src/main/java/fun/cosmozhu/session4/StreamTest.java

作者:cosmozhu --90后的老父亲,专注于保护地球的程序员

个人网站:https://www.cosmozhu.fun

欢迎转载,转载时请注明出处。

相关文章

本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 需求
  • 解决方案
  • 执行效果
  • 代码地址
    • 相关文章
    相关产品与服务
    对象存储
    对象存储(Cloud Object Storage,COS)是由腾讯云推出的无目录层次结构、无数据格式限制,可容纳海量数据且支持 HTTP/HTTPS 协议访问的分布式存储服务。腾讯云 COS 的存储桶空间无容量上限,无需分区管理,适用于 CDN 数据分发、数据万象处理或大数据计算与分析的数据湖等多种场景。
    领券
    问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档