• Flink系列文档-(YY04)-Flink编程基础API-Transformation算子

    大多处理数据的流程是   1)加载数据源数据   2)转换处理  3) 输出结果 

    1 映射类算子

    1.1 map算子

    map(new MapFunction )

    MapFunction: （x）-> y   [1条变1条]

    1. /**
    2.  * @Date: 22.11.8
    3.  * @Author: Hang.Nian.YY
    4.  * @qq: 598196583
    5.  * @Tips: 学大数据 ,到多易教育
    6.  * @Description:
    7.  */
    8. public class Base_API_MapFunction {
    9.     public static void main(String[] args) throws Exception {
    10.  
    11.         Configuration conf = new Configuration();
    12.         conf.setInteger("rest.port", 8888);
    13.         StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
    14.         see.setParallelism(1);
    15.  
    16.         // 加载网络数据流
    17.         DataStreamSource ds = see.socketTextStream("linux01", 9999);
    18.  
    19.         // 使用转换算子 map 处理数据  一条一条的处理数据
    20.  
    21.         /**
    22.          * 示例一
    23.          */
    24.         SingleOutputStreamOperator sos = ds.map(new MapFunction() {
    25.             // 每条数据调用一次
    26.             @Override
    27.             public String map(String line) throws Exception {
    28.                 // 将每条数据转换成大写
    29.                 return line.toUpperCase();
    30.             }
    31.         });
    32.         /**
    33.          * 示例二
    34.          */
    35.         SingleOutputStreamOperator> sos2 = ds.map(new MapFunction>() {
    36.             @Override
    37.             public Tuple2 map(String line) throws Exception {
    38.                 // 将接收的数据 封装成二元组
    39.                 String[] split = line.split("\\s+");
    40.                 Tuple2 tp2 = Tuple2.of(split[0], split[1]);
    41.                 return tp2;
    42.             }
    43.         });
    44.  
    45.         /**
    46.          * 示例三
    47.          * MapFunction 接口中只有一个抽象各个  可以使用Lamda表达式的方式处理数据
    48.          *   public interface MapFunction extends Function, Serializable {
    49.          *         O map (T value) throws Exception;
    50.          *    }
    51.          *}
    52.          */
    53.         SingleOutputStreamOperator> sos3=  ds.map(line->{
    54.             String[] arr = line.split("\\s+");
    55.             return  Tuple2.of(arr[0] , arr[1]) ;
    56.         }) .returns(TypeInformation.of(new TypeHint>() {})) ; // 指定返回值的数据类型
    57.            // 或者 .returns(new TypeHint>() {})  ; // 指定返回值的数据类型
    58.         sos3.print("map后的数据: ") ;
    59.         see.execute() ;
    60.         
    61.     }
    62. }

    如果是调用map方法时传入Lambda表达式，需要在调用map方法后，在调用returns方法指定返回的数据的类型。不然Flink无法自动推断出返回的数据类型，会出现异常。

    1.2 flatMap扁平映射 

    flatMap( new FlatMapFcuntion)

    FlatMapFunction: x-> x1, x2,x3,x4  [1条变多条，并展平]

    1. /**
    2.  * @Date: 22.11.8
    3.  * @Author: Hang.Nian.YY
    4.  * @qq: 598196583
    5.  * @Tips: 学大数据 ,到多易教育
    6.  * @Description:
    7.  */
    8. public class Base_API_FlatMapFunction {
    9.     public static void main(String[] args) throws Exception {
    10.  
    11.         Configuration conf = new Configuration();
    12.         conf.setInteger("rest.port", 8888);
    13.         StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
    14.         see.setParallelism(1);
    15.         // 加载网络数据流
    16.         DataStreamSource ds = see.socketTextStream("linux01", 9999);
    17.  
    18.         /**
    19.          * 示例一
    20.          * 将接收到的一行数据  扁平化处理
    21.          * 组装成  (单词和1)
    22.          */
    23.         SingleOutputStreamOperator> soo = ds.flatMap(new FlatMapFunction>() {
    24.             @Override
    25.             public void flatMap(String line, Collector> out) throws Exception {
    26.                 String[] arr = line.split("\\s+");
    27.                 for (String word : arr) {
    28.                     Tuple2 tp = Tuple2.of(word, 1);
    29.                     // 接收到一条数据 ,将一条数据转成成 多条数据后  使用Collector 收集多条数据
    30.                     out.collect(tp);
    31.                 }
    32.             }
    33.         });
    34.  
    35.         /**
    36.          * 示例二
    37.          *   使用lambda 表达式处理数据
    38.          *   不会自动推断返回值数据 类型  可以使用returns 指定返回值数据类型
    39.          */
    40.         SingleOutputStreamOperator> soo2 = ds.flatMap((String line, Collector> out) -> {
    41.             String[] arr = line.split("\\s+");
    42.             for (String s : arr) {
    43.                 out.collect(Tuple2.of(s, 1));
    44.             }
    45.         }).returns(Types.TUPLE(Types.STRING, Types.INT));
    46.         soo2.print("扁平化后的数据: ");
    47.         see.execute("flatMap函数示例") ;
    48.         
    49.     }
    50. }

    1.3 project 投影

    该算子只能对Tuple类型数据使用，project方法的功能类似sql中的"select 字段"；

    该方法只有Java的API有，Scala的API没此方法。

    1. /**
    2.  * @Date: 22.11.8
    3.  * @Author: Hang.Nian.YY
    4.  * @qq: 598196583
    5.  * @Tips: 学大数据 ,到多易教育
    6.  * @Description:
    7.  */
    8. public class _3Base_API_Project {
    9.     public static void main(String[] args) throws Exception {
    10.  
    11.         Configuration conf = new Configuration();
    12.         conf.setInteger("rest.port", 8888);
    13.         StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
    14.         see.setParallelism(1);
    15.  
    16.         /**
    17.          * project 投影（DataStream → DataStream）
    18.          * 该算子只能对Tuple类型数据使用，project方法的功能类似sql中的"select 字段"；
    19.          * 该方法只有Java的API有，Scala的API没此方法。
    20.          */
    21.         DataStreamSource> ds = see.fromElements(
    22.                 Tuple4.of(1, "YY", "F", 100),
    23.                 Tuple4.of(2, "DY", "F", 99)
    24.  
    25.         );
    26.         //  处理每条数据  ,返回每条数据中的指定位置的属性值  
    27.         // 只要 id 和 name 
    28.         SingleOutputStreamOperator res = ds.project(0, 1);
    29.         res.print() ;
    30.         see.execute("project函数") ;
    31.         
    32.     }
    33. }

    2 过滤算子

      filter过滤（DataStream → DataStream）

    filter(new FilterFunction)

    FilterFunction :  x -> true/false

    1. /**
    2.  * @Date: 22.11.8
    3.  * @Author: Hang.Nian.YY
    4.  * @qq: 598196583
    5.  * @Tips: 学大数据 ,到多易教育
    6.  * @Description:
    7.  */
    8. public class _4Base_API_FilterFunction {
    9.  
    10.     public static void main(String[] args) throws Exception {
    11.         Configuration conf = new Configuration();
    12.         conf.setInteger("rest.port", 8888);
    13.         StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
    14.         see.setParallelism(1);
    15.  
    16.         // 获取数据源
    17.         DataStreamSource ds = see.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9);
    18.         /**
    19.          *  过滤出偶数
    20.          */
    21.         SingleOutputStreamOperator res = ds.filter(new FilterFunction() {
    22.             @Override
    23.             public boolean filter(Integer value) throws Exception {
    24.                 return value % 2 == 0;
    25.             }
    26.         });
    27.         //使用 lambda表达式   过滤出奇数
    28.         SingleOutputStreamOperator res2 = ds.filter(e -> {
    29.             return e % 2 == 1;
    30.         });
    31.  
    32.         /**
    33.          * 示例 ----- 
    34.          * 过滤出分数大于60的人
    35.          */
    36.         DataStreamSource yyds = see.fromElements(
    37.                 new YY(1, "DY", 100),
    38.                 new YY(2, "XY", 100),
    39.                 new YY(3, "HH", 10),
    40.                 new YY(4, "XH", 12)
    41.         );
    42.         SingleOutputStreamOperator res3 = yyds.filter(new FilterFunction() {
    43.             @Override
    44.             public boolean filter(YY yy) throws Exception {
    45.                 return yy.getScore() > 60;
    46.             }
    47.         });
    48.  
    49.         res3.print();
    50.         see.execute("filter function");
    51.  
    52.     }
    53. }
    54.  
    55. @Data
    56. @NoArgsConstructor
    57. @AllArgsConstructor
    58. @ToString
    59. class YY {
    60.     private int id;
    61.     private String name;
    62.     private double score;
    63. }

    3 分组算子 

     keyBy按key分组（DataStream → KeyedStream）

    1. /**
    2.  * @Date: 22.11.8
    3.  * @Author: Hang.Nian.YY
    4.  * @qq: 598196583
    5.  * @Tips: 学大数据 ,到多易教育
    6.  * @Description:
    7.  */
    8. public class _5Base_API_KeyBy {
    9.     public static void main(String[] args) throws Exception {
    10.  
    11.         Configuration conf = new Configuration();
    12.         conf.setInteger("rest.port", 8888);
    13.         StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
    14.         see.setParallelism(1);
    15.         // 加载网络数据流
    16.         DataStreamSource ds = see.socketTextStream("linux01", 9999);
    17.  
    18.         SingleOutputStreamOperator> wordAndOne = ds.flatMap(new FlatMapFunction>() {
    19.             @Override
    20.             public void flatMap(String line, Collector> out) throws Exception {
    21.                 String[] arr = line.split("\\s+");
    22.                 for (String word : arr) {
    23.                     Tuple2 tp = Tuple2.of(word, 1);
    24.                     // 接收到一条数据 ,将一条数据转成成 多条数据后  使用Collector 收集多条数据
    25.                     out.collect(tp);
    26.                 }
    27.             }
    28.         });
    29.  
    30.         /**
    31.          * 对数据流进行分组 
    32.          *    -- 按照单词分组
    33.          */
    34.         
    35.         // 按照单词分组
    36.         wordAndOne.keyBy(0) ;
    37.         // 按照单词分组   KeyedStream
    38.         KeyedStream, String> res = wordAndOne.keyBy(new KeySelector, String>() {
    39.             @Override
    40.             public String getKey(Tuple2 value) throws Exception {
    41.                 return value.f0;
    42.             }
    43.         });
    44.         //根据自定义数据类型中的某个属性进行分组 
    45.         DataStreamSource ds2 = see.fromElements(
    46.                 new YY2(1, "DY", "NM_BT", 100),
    47.                 new YY2(2, "XY", "NM_BT", 100),
    48.                 new YY2(3, "HH", "SD_HZ", 10),
    49.                 new YY2(4, "XH", "SD_HZ", 12)
    50.         );
    51.  
    52.         /**
    53.          * 对数据流进行分组 
    54.          *    -- 根据Bean的属性 
    55.          */
    56.         ds2.keyBy(new KeySelector() {
    57.             @Override
    58.             public String getKey(YY2 value) throws Exception {
    59.                 return value.getCity();
    60.             }
    61.         }) ;
    62.         
    63.         ds2.keyBy(YY2::getCity) ;
    64.         res.print() ;
    65.         see.execute() ;
    66.         
    67.     }
    68. }
    69.  
    70. @Data
    71. @NoArgsConstructor
    72. @AllArgsConstructor
    73. @ToString
    74. class YY2 {
    75.     private int id;
    76.     private String name;
    77.     private  String  city ;
    78.     private double score;
    79. }

    4 滚动聚合算子

    1. 此处所说的滚动聚合算子，是多个聚合算子的统称，有sum、min、minBy、max、maxBy；
    2. 这些算子的底层逻辑都是维护一个聚合值，并使用每条流入的数据对聚合值进行滚动更新；
    3. 这些算子都只能在KeyedStream上调用（就是必须keyby后调用）；

     4.1 sum

    1. package com.blok;
    2.  
    3. import lombok.AllArgsConstructor;
    4. import lombok.Data;
    5. import lombok.NoArgsConstructor;
    6. import lombok.ToString;
    7. import org.apache.flink.api.common.functions.FlatMapFunction;
    8. import org.apache.flink.api.java.functions.KeySelector;
    9. import org.apache.flink.api.java.tuple.Tuple2;
    10. import org.apache.flink.configuration.Configuration;
    11. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    12. import org.apache.flink.streaming.api.datastream.KeyedStream;
    13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    15. import org.apache.flink.util.Collector;
    16.  
    17. /**
    18.  * @Date: 22.11.8
    19.  * @Author: Hang.Nian.YY
    20.  * @qq: 598196583
    21.  * @Tips: 学大数据 ,到多易教育
    22.  * @Description:
    23.  */
    24. public class _6Base_API_Sum{
    25.     public static void main(String[] args) throws Exception {
    26.  
    27.         Configuration conf = new Configuration();
    28.         conf.setInteger("rest.port", 8898);
    29.         StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
    30.         see.setParallelism(1);
    31.         // 加载网络数据流
    32.         DataStreamSource ds = see.socketTextStream("linux01", 9999);
    33.  
    34.         SingleOutputStreamOperator> wordAndOne = ds.flatMap(new FlatMapFunction>() {
    35.             @Override
    36.             public void flatMap(String line, Collector> out) throws Exception {
    37.                 String[] arr = line.split("\\s+");
    38.                 for (String word : arr) {
    39.                     Tuple2 tp = Tuple2.of(word, 1);
    40.                     // 接收到一条数据 ,将一条数据转成成 多条数据后  使用Collector 收集多条数据
    41.                     out.collect(tp);
    42.                 }
    43.             }
    44.         });
    45.  
    46.         /**
    47.          * 对数据流进行分组
    48.          *    -- 按照单词分组
    49.          */
    50.  
    51.         // 按照单词分组
    52.         wordAndOne.keyBy(0) ;
    53.         // 按照单词分组   KeyedStream
    54.         KeyedStream, String> res = wordAndOne.keyBy(new KeySelector, String>() {
    55.             @Override
    56.             public String getKey(Tuple2 value) throws Exception {
    57.                 return value.f0;
    58.             }
    59.         });
    60.         //------------------------------------------------------------------------------
    61.          // 滚动聚合: 随着数据的流出 结果数据源源的进行数据叠加
    62.         // 统计单词出现的次数
    63.         SingleOutputStreamOperator> sum = res.sum("1");
    64.         SingleOutputStreamOperator> sum2 = res.sum(1);
    65.  
    66.         //------------------------------------------------------------------------------
    67.  
    68.         //根据自定义数据类型中的某个属性进行分组
    69.         DataStreamSource ds2 = see.fromElements(
    70.                 new YY2(1, "DY", "NM_BT", 100),
    71.                 new YY2(2, "XY", "NM_BT", 100),
    72.                 new YY2(3, "HH", "SD_HZ", 10),
    73.                 new YY2(4, "XH", "SD_HZ", 12)
    74.         );
    75.  
    76.         /**
    77.          * 对数据流进行分组
    78.          *    -- 根据Bean的属性
    79.          */
    80.         KeyedStream keyed = ds2.keyBy(new KeySelector() {
    81.             @Override
    82.             public String getKey(YY2 value) throws Exception {
    83.                 return value.getCity();
    84.             }
    85.         });
    86.  
    87.         //------------------------------------------------------------------------------
    88.         // 滚动聚合: 随着数据的流出 结果数据源源的进行数据叠加
    89.         // 统计每组的总分  根据组内Bean的属性
    90.         SingleOutputStreamOperator score = keyed.sum("score");
    91.         score.print() ;
    92.         //------------------------------------------------------------------------------
    93.         see.execute() ;
    94.  
    95.     }
    96. }

    4.2 min/minBy/max/maxBy

    这两个算子都是求最小值；min和minBy的区别在于：

    1. min的返回值，最小值字段以外，其他字段是第一条输入数据的值；
    2. minBy返回值，就是最小值字段所在的那条数据；

    底层原理：滚动更新时是更新一个字段，还是更新整条数据的区别；

    1. package com.blok;
    2.  
    3. import org.apache.flink.api.common.functions.FlatMapFunction;
    4. import org.apache.flink.api.common.functions.MapFunction;
    5. import org.apache.flink.api.java.functions.KeySelector;
    6. import org.apache.flink.api.java.tuple.Tuple2;
    7. import org.apache.flink.configuration.Configuration;
    8. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    9. import org.apache.flink.streaming.api.datastream.KeyedStream;
    10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    12. import org.apache.flink.util.Collector;
    13.  
    14. /**
    15.  * @Date: 22.11.8
    16.  * @Author: Hang.Nian.YY
    17.  * @qq: 598196583
    18.  * @Tips: 学大数据 ,到多易教育
    19.  * @Description:
    20.  */
    21. public class _7Base_API_MaxMin {
    22.     public static void main(String[] args) throws Exception {
    23.  
    24.         Configuration conf = new Configuration();
    25.         conf.setInteger("rest.port", 8898);
    26.         StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
    27.         see.setParallelism(1);
    28.         // 加载网络数据流
    29.         DataStreamSource ds = see.socketTextStream("linux01", 9999);
    30.  
    31.         /**
    32.          * public class YY2 {
    33.          *     private int id;
    34.          *     private String name;
    35.          *     private  String  city ;
    36.          *     private double score;
    37.          * }
    38.          */
    39.         SingleOutputStreamOperator beans = ds.map(new MapFunction() {
    40.             @Override
    41.             public YY2 map(String value) throws Exception {
    42.                 String[] arr = value.split(",");
    43.                 YY2 yy = new YY2(Integer.parseInt(arr[0]), arr[1], arr[2], Double.parseDouble(arr[3]));
    44.                 return yy;
    45.             }
    46.         });
    47.         //将 数据按照城市分组
    48.         KeyedStream keyed = beans.keyBy(YY2::getCity);
    49.         //---------------------------------------------------------------
    50.         /**
    51.          * min 返回的是第一条数据 但是  会修改第一条数据 指定的属性信息
    52.          * max 返回的是第一条数据 但是  会修改第一条数据 指定的属性信息
    53.          *
    54.          * 分数最小的数据: 1,yy1,NM_BT,98
    55.          * 分数最大的数据: 1,yy1,NM_BT,100
    56.          */
    57.         // 获取分数最低的 信息
    58.         SingleOutputStreamOperator minScoreInfo = keyed.min("score");
    59.         SingleOutputStreamOperator maxScoreInfo = keyed.max("score");
    60.         /**
    61.          * 测试数据
    62.          * 1,yy1,NM_BT,99
    63.          * 2,yy2,NM_BT,100
    64.          * 3,yy3,NM_BT,98
    65.          * 4,yy4,NM_BT,98.5
    66.          * 1,hh1,SD_HZ,99
    67.          * 2,hh2,SD_HZ,100
    68.          * 3,hh3,SD_HZ,98
    69.          * 4,hh4,SD_HZ,98.5
    70.          * 5,hh5,SD_HZ,101
    71.          */
    72.         //---------------------------------------------------------------
    73.         /**
    74.          * maxBy  minBy返回的就是那条指定属性最大(最小)的数据
    75.          *
    76.          * 分数最大的数据: > YY2(id=2, name=yy2, city=NM_BT, score=100.0)
    77.          * 分数最小的数据: > YY2(id=3, name=yy3, city=NM_BT, score=98.0)
    78.          */
    79.         SingleOutputStreamOperator minScoreInfoBy = keyed.minBy("score");
    80.         SingleOutputStreamOperator maxScoreInfoBy = keyed.maxBy("score");
    81.         minScoreInfoBy.print("分数最小的数据: ") ;
    82.         maxScoreInfoBy.print("分数最大的数据: ") ;
    83.  
    84.         see.execute() ;
    85.  
    86.     }
    87. }

    4.3 reduce

    它的滚动聚合逻辑没有写死，而是由用户通过ReduceFunction来传入。

    1. /**
    2.  * @Date: 22.11.8
    3.  * @Author: Hang.Nian.YY
    4.  * @qq: 598196583
    5.  * @Tips: 学大数据 ,到多易教育
    6.  * @Description:
    7.  */
    8. public class _8Base_API_Reduce {
    9.     public static void main(String[] args) throws Exception {
    10.  
    11.         Configuration conf = new Configuration();
    12.         conf.setInteger("rest.port", 8898);
    13.         StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
    14.         see.setParallelism(1);
    15.         // 加载网络数据流
    16.         DataStreamSource ds = see.socketTextStream("linux01", 9999);
    17.  
    18.         /**
    19.          * public class YY2 {
    20.          *     private int id;
    21.          *     private String name;
    22.          *     private  String  city ;
    23.          *     private double score;
    24.          * }
    25.          */
    26.         SingleOutputStreamOperator beans = ds.map(new MapFunction() {
    27.             @Override
    28.             public YY2 map(String value) throws Exception {
    29.                 String[] arr = value.split(",");
    30.                 YY2 yy = new YY2(Integer.parseInt(arr[0]), arr[1], arr[2], Double.parseDouble(arr[3]));
    31.                 return yy;
    32.             }
    33.         });
    34.  
    35.         // 将数据beans分组
    36.         KeyedStream keyed = beans.keyBy(YY2::getCity);
    37.         SingleOutputStreamOperator reduced = keyed.reduce(new ReduceFunction() {
    38.             @Override
    39.             public YY2 reduce(YY2 value1, YY2 value2) throws Exception {
    40.                 YY2 yy2 = new YY2();
    41.                 yy2.setScore(value1.getScore() + value2.getScore());
    42.                 yy2.setCity(value1.getCity());
    43.                 return yy2;
    44.             }
    45.         });
    46.         reduced.print("聚合后的结果");
    47.         see.execute() ;
    48.  
    49.     }
    50. }

  • 相关阅读:
    标签类目体系（面向业务的数据资产设计方法论）-读书笔记1
    使用VBA快速完成不规则数据整理
    14、JAVA入门——方法和构造方法
    【小题练手】---Java基础
    职场小白如何将图片转文字？这个方法建议收藏使用！
    Kotlin高仿微信-第2篇-登录
    PendingIntent
    Ant-Design-Pro-V5 ：ProTable自定义搜索菜单操作栏和搜索事件、列表工具栏操作。
    计算机网络原理 运输层
    Android组件化神器 --- ARouter
  • 原文地址：https://blog.csdn.net/qq_37933018/article/details/127741870