2021年最新最全Flink系列教程_Flink流批一体API

Posted ChinaManor

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了2021年最新最全Flink系列教程_Flink流批一体API相关的知识,希望对你有一定的参考价值。

day02-03_流批一体API

今日目标

  • 流处理原理初探

  • 流处理概念(理解)

  • 程序结构之数据源Source(掌握)

  • 程序结构之数据转换Transformation(掌握)

  • 程序结构之数据落地Sink(掌握)

  • Flink连接器Connectors(理解)

流处理原理初探

  • Flink的角色分配

    1. JobMaster 老大, 主要负责 集群的管理, 故障的恢复, checkpoint 检查点设置
    2. taskmanager worker 小弟, 具体负责任务的执行节点
    3. client 提交任务的界面
  • taskmanager 执行能力

    1. taskslot 静态的概念
    2. parallelism 并行度 动态概念

    image-20210616083826956

  • 每个节点就是一个 task 任务

    每个任务拆分成多个并行处理的任务, 多个线程就有多个子任务,就叫子任务 subtask

  • 流图 StreamGraph 逻辑执行流图 DataFlow

    operator chain 操作链

  • JobGraph

    ExecuteGraph 物理执行计划

  • Event 事件 带有时间戳的

  • Operator 传递模式 : one to one 模式, redistributing模式

image-20210504173717509

  • Flink之执行图

image-20210504174448457

流处理概念

数据的时效性

  • 强调的是数据的处理时效

    处理的时间窗口, 按月, 按天, 按小时还是秒级处理

流处理和批处理

  • 批处理是有界的数据

    • 处理完整的数据集, 比如排序数据, 计算全局的状态, 生成最终的输入概述.
    • 批量计算: 统一收集数据->存储到DB->对数据进行批量处理
  • 流处理是无界的数据

    • 窗口操作来划分数据的边界进行计算
    • 流式计算,顾名思义,就是对数据流进行处理
  • 在Flink1.12时支持流批一体 既支持流处理也支持批处理。

    image-20210505091439155

  • 流批一体 Flink1.12.x 批处理和流处理

    • 可复用性: 作业在流模式或者批处理两种模式自由切换, 无需重写任何代码.
    • 维护简单: 统一的 API 意味着流和批可以共用同一组 connector,维护同一套代码.

编程模型

image-20210505091654525

  • source - 读取数据源
  • transformation - 数据转换 map flatMap groupBy keyBy sum
  • sink - 落地数据 addSink print

Source

基于集合的Source

  • 开发和测试使用

  • 分类

    1.env.fromElements(可变参数);
    2.env.fromColletion(各种集合);
    # 过期
    3.env.generateSequence(开始,结束);
    4.env.fromSequence(开始,结束);
    
  • 使用集合 Source

    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    import java.util.ArrayList;
    
    /**
     * Author itcast
     * Date 2021/6/16 9:29
     * 需求: 通过集合source打印结果,查看如何使用
     * 开发步骤:
     * 1. 创建流环境
     * 2. 从集合中读取数据
     * 3. 打印输出
     * 4. 运行执行
     */
    public class SourceDemo01 {
        public static void main(String[] args) throws Exception {
            //1. 创建流环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //2. 从集合中读取数据
            //2.1 fromElement 从元素集合
            DataStreamSource<String> source1 = env.fromElements("hello world", "hello spark", "hello flink");
            //2.2 fromCollection  从集合列表
            ArrayList<String> strings = new ArrayList<>();
            strings.add("hello world");
            strings.add("hello flink");
            DataStreamSource<String> source2 = env.fromCollection(strings);
            //2.3 fromSequence   从序列
            DataStreamSource<Long> source3 = env.fromSequence(1, 10);
            //3. 打印输出
            source1.print();
            //4. 运行执行
            env.execute();
        }
    }
    
  • socket 数据源 wordcount 统计

    /**
     * Author itcast
     * Desc
     * SocketSource
     */
    public class SourceDemo03 {
        public static void main(String[] args) throws Exception {
            //1.env
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
            //2.source
            DataStream<String> linesDS = env.socketTextStream("node1", 9999);
    
            //3.处理数据-transformation
            //3.1每一行数据按照空格切分成一个个的单词组成一个集合
            DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {
                @Override
                public void flatMap(String value, Collector<String> out) throws Exception {
                    //value就是一行行的数据
                    String[] words = value.split(" ");
                    for (String word : words) {
                        out.collect(word);//将切割处理的一个个的单词收集起来并返回
                    }
                }
            });
            //3.2对集合中的每个单词记为1
            DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public Tuple2<String, Integer> map(String value) throws Exception {
                    //value就是进来一个个的单词
                    return Tuple2.of(value, 1);
                }
            });
    
            //3.3对数据按照单词(key)进行分组
            //KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOnesDS.keyBy(0);
            KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(t -> t.f0);
            //3.4对各个组内的数据按照数量(value)进行聚合就是求sum
            DataStream<Tuple2<String, Integer>> result = groupedDS.sum(1);
    
            //4.输出结果-sink
            result.print();
    
            //5.触发执行-execute
            env.execute();
        }
    }
    
  • 自定义数据源 - 随机数据

    import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
    
    import java.util.Random;
    import java.util.UUID;
    
    /**
     * Author itcast
     * Date 2021/6/16 10:18
     * 每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
     * 要求:
     * - 随机生成订单ID(UUID)
     * - 随机生成用户ID(0-2)
     * - 随机生成订单金额(0-100)
     * - 时间戳为当前系统时间
     *
     * SourceFunction:非并行数据源(并行度只能=1)
     * RichSourceFunction:多功能非并行数据源(并行度只能=1)
     * ParallelSourceFunction:并行数据源(并行度能够>=1)
     * RichParallelSourceFunction:多功能并行数据源(并行度能够>=1)--后续学习的Kafka数据源使用的就是该接口
     */
    public class CustomSource01 {
        public static void main(String[] args) throws Exception {
            //1.env 创建 StreamExectution
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            //2.source,创建自动生成 Order 数据源
            DataStreamSource<Order> source = env.addSource(new MyOrderSource());
            //3.打印数据源
            source.print();
            //4.执行
            env.execute();
            //定义实体类 Order 包括四个字段 oid uid money currentTime
            //定义静态内部类 MyOrderSource 继承 RichParallelSourceFunction
            //每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
            //要求:
            //- 随机生成订单ID(UUID)
            //- 随机生成用户ID(0-2)
            //- 随机生成订单金额(0-100)
            //- 时间戳为当前系统时间
        }
    
        public static class MyOrderSource extends RichParallelSourceFunction<Order> {
            boolean flag = true;
            Random rn = new Random();
            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                //每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
                //要求:
                while(flag) {
                    //- 随机生成订单ID(UUID)
                    String oid = UUID.randomUUID().toString();
                    //- 随机生成用户ID(0-2)
                    int uid = rn.nextInt(3);
                    //- 随机生成订单金额(0-100)
                    int money = rn.nextInt(101);
                    //- 时间戳为当前系统时间
                    long currentTime = System.currentTimeMillis();
                    ctx.collect(new Order(oid,uid,money,currentTime));
                    //一秒钟休息一下
                    Thread.sleep(1000);
                }
            }
    
            @Override
            public void cancel() {
                flag = false;
            }
        }
        //创建 Order 对象
        @AllArgsConstructor
        @NoArgsConstructor
        @Data
        public static class Order{
            private String oid;
            private int uid;
            private int money;
            private long currentTime;
        }
    }
    
  • 自定义数据源 - 从mysql数据库中读取 t_student表中数据( 这种场景用的非常少 - 自定义数据源 )

    import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
    
    import java.sql.Connection;
    import java.sql.DriverManager;
    import java.sql.PreparedStatement;
    import java.sql.ResultSet;
    
    /**
     * Author itcast
     * Date 2021/6/16 10:37
     * 需求:
     * 从MySql数据库中读取 t_student 表数据
     * 开发步骤:
     * //1.env 设置并行度为 1
     * //2.source,创建连接MySQL数据源 数据源,每2秒钟生成一条数据
     * //3.打印数据源
     * //4.执行
     * //创建静态内部类 Student ,字段为 id:int name:String age:int
     * //创建静态内部类 MySQLSource 继承RichParallelSourceFunction<Student>
     * // 实现 open 方法 ,创建 connection 和 prepareStatement
     * // 获取数据库连接 mysql5.7版本
     *  jdbc:mysql://192.168.88.163:3306/bigdata?useSSL=false
     * // 实现 run 方法, 每 5秒钟创建一条数据
     * // 实现 close 方法
     */
    public class CustomSourceMySQL {
        public static void main(String[] args) throws Exception {
            //1.env 设置并行度为 1
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            //2.source,创建连接MySQL数据源 数据源,每2秒钟生成一条数据
            DataStreamSource<Student> source = env.addSource(new MySQLSource());
            //3.打印数据源
            source.print();
            //4.执行
            env.execute();
            //创建静态内部类 Student ,字段为 id:int name:String age:int
            //创建静态内部类 MySQLSource 继承RichParallelSourceFunction<Student>
            // 实现 open 方法 ,创建 connection 和 prepareStatement
            // 获取数据库连接 mysql5.7版本
             jdbc:mysql://192.168.88.163:3306/bigdata?useSSL=false
            // 实现 run 方法, 每 5秒钟创建一条数据
            // 实现 close 方法
        }
        public static class MySQLSource extends RichSourceFunction<Student> {
            boolean flag = true;
            Connection conn = null;
            PreparedStatement ps = null;
            // open 生命周期的开始, 只做一次
            @Override
            public void open(Configuration parameters) throws Exception {
                conn = DriverManager.getConnection("jdbc:mysql://192.168.88.163:3306/bigdata?useSSL=false",
                        "root", "123456");
                String sql = "select id,name,age from t_student";
                // 执行 preparement
                ps = conn.prepareStatement(sql);
            }
    
            @Override
            public void run(SourceContext<Student> ctx) throws Exception {
                while(flag){
                    // 查询结果集
                    ResultSet rs = ps.executeQuery();
                    while(rs.next()){
                        int id = rs.getInt("id");
                        String name = rs.getString("name");
                        int age = rs.getInt("age");
                        ctx.collect(new Student(id,name,age));
                        Thread.sleep(5000);
                    }
                }
            }
    
            @Override
            public void cancel() {
                flag = false;
            }
    
            //关闭数据库, 在整个生命周期也只做一次
            @Override
            public void close() throws Exception {
                if(!ps.isClosed()) ps.close();
                if(!conn.isClosed()) conn.close();
            }
        }
        //定义 student
        @AllArgsConstructor
        @NoArgsConstructor
        @Data
        public static class Student{
            private int id;
            private String name ;
            private int age;
        }
    }
    

合并-拆分

  • 合并数据流 将两个数据流合并成一个数据流

  • 应用场景

    ① 将不同的数据源 电脑, app , ipad ,微信小程序的所有的订单的信息 ,统计分析,挖掘

    ② 将不同的手持设备, 电脑用户行为轨迹收集统计分析

  • union 和connect 区别

    union 和 connect 合流

    union 算子 要求数据流的类型必须保持一致.

    connect 算子 要求数据流的类型可以不一致

  • 需求: 将两个数据流合并到一起

    /**
     * Author itcast
     * Desc
     */
    public class TransformationDemo02 {
        public static void main(String[] args) throws Exception {
            //1.env
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
            env.setParallelism(1);
    
            //2.Source
            DataStream<String> ds1 = env.fromElements("hadoop", "spark", "flink");
            DataStream<String> ds2 = env.fromElements("oozie", "flume", "flink");
            DataStream<Long> ds3 = env.fromElements(1L, 2L, 3L);
    
            //3.Transformation
            // union 算子 保证两个数据流类型保持一致
            DataStream<String> result1 = ds1.union(ds2);//合并但不去重 https://blog.csdn.net/valada/article/details/104367378
            // connect 算子 两个数据流类型可以不一样
            ConnectedStreams<String, Long> tempResult = ds1.connect(ds3);
            //interface CoMapFunction<IN1, IN2, OUT>
            DataStream<String> result2 = tempResult.map(new CoMapFunction<String, Long, String>() {
                @Override
                public String map1(String value) throws Exception {
                    return "String->String:" + value;
                }
    
                @Override
                public String map2(Long value) throws Exception {
                    return "Long->String:" + value.toString();
                }
            });
    
            //4.Sink
            //result1.print();
            result2.print();
    
            //5.execute
            env.execute();
        }
    }
    

分流 select 和 outputside

  • 将一个数据流分成多个数据流

  • 应用场景

    ① 服务器日志 分流出正常的日志, 告警日志, 报错日志

  • 需求 - 将数据流拆分成 偶数 和 奇数

  • 开发步骤

    import org.apache.flink.api.common.typeinfo.TypeInformation;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.ProcessFunction;
    import org.apache.flink.util.Collector;
    import org.apache.flink.util.OutputTag;
    
    /**
     * Author itcast
     * Date 2021/6/16 11:29
     * 需求: 拆分数据
     * 开发步骤:
     * //1.env
     * //2.Source 比如 1-20之间的数字
     * //定义两个输出tag 一个奇数 一个偶数,指定类型为Long
     * //对source的数据进行process处理区分奇偶数
     * //3.获取两个侧输出流
     * //4.sink打印输出
     * //5.execute
     */
    public class SplitDataStream {
        public static void main(String[] args) throws Exception {
            //1.env
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            //2.Source 比如 1-20之间的数字
            DataStreamSource<Long> source = env.fromSequence(1, 21);
            //定义两个侧输出流 一个奇数 一个偶数, 指定类型为Long
            // 需要指定对应的数据类型, 默认OutputTag 会用通用类型,需要手动
            OutputTag<Long> odd = new OutputTag<Long>("odd", TypeInformation.of(Long.class));
            OutputTag<Long> even = new OutputTag<Long>("even", TypeInformation.of(Long.class));
            //对source的数据进行process处理区分奇偶数
            SingleOutputStreamOperator<Long> result = source.process(new ProcessFunction<Long, Long>() {
                @Override
                public void processElement(Long value, Context ctx, Collector<

    以上是关于2021年最新最全Flink系列教程_Flink流批一体API的主要内容,如果未能解决你的问题,请参考以下文章

    2021年最新最全Flink系列教程_Flink原理初探和流批一体API

    2021年最新最全Flink系列教程_Flink快速入门(概述,安装部署)(建议收藏!!)

    2021年最新最全Flink系列教程_Flink原理初探和流批一体API(二.五)

    2021年最新最全Flink系列教程__Flink高级API

    收藏+下载!Flink 社区 2021 最新最全学习渠道汇总

    收藏+下载!Flink 社区 2021 最新最全学习渠道汇总