Flink入门

淡淡的烟草味﹌ 2022-11-12 01:39 218阅读 0赞

## 一、Flink是什么 ##

Apache Flink是一个**框架和分布式**处理引擎，用于对**无界**和**有界**数据流进行**状态计算**

## 二、为什么选择Flink ##

流数据能够反映我们真实的生活方式，但是传统的数据架构是基于有限数据集的。

我们期望的数据处理框架是：（1）低延迟（2）高吞吐（3）结果的准确性和良好的容错性

**图一：传统事务处理架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]  
**图二：传统分析处理架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]  
**图三：有状态的流式处理**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 2]  
**图四：lambda流处理架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 3]  
**图五：Flink事件驱动架构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 4]  
**Flink vs Spark Streaming**

1、数据模型

*  spark采用RDD模型，spark streaming的DStream实际上也就是一组组小批数据 RDD 的集合
 *  flink基本数据模型是数据流，以及事件（Event）序列

2、运行时架构

*  spark是批计算，将DAG划分为不同的stage，一个stage完成后才可以计算下一个
 *  flink是标准的流执行模式，一个事件在一个节点处理完后可以直接发往下一个节点进行处理

**Flink的特点总结：**

*  低延迟，每秒处理数百万个事件，毫秒级延迟
 *  高吞吐
 *  结果的准确性和良好的容错性
 *  支持事件时间和处理时间
 *  精确一次的状态一致性保证
 *  与众多常用存储系统的连接
 *  高可用，动态扩展，实现 7✖️24 小时全天候运行

## 三、示例代码 ##

### 3.1 批处理单词统计demo ###

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.test.sun</groupId>
        <artifactId>FirstFlinkTest</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            <flink.version>1.10.1</flink.version>
        </properties>
    
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.2.0.RELEASE</version>
            <relativePath/>
        </parent>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.12</artifactId>
                <version>${flink.version}</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-java</artifactId>
                <version>${flink.version}</version>
            </dependency>
        </dependencies>
    </project>

hello.txt

sunxiaoke hello demo
    first flink demo
    just do it
    flink flink flink

WordCount类

package com.test.sun.demo;
    
    import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.DataSet;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.util.Collector;
    
    /**
     * @description: 批处理Word Count
     * @time: 2021-03-21 19:20
     */
    public class WordCount { 
        public static void main(String[] args) throws Exception{ 
            // 创建执行环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            // 从文件中读取数据
            String inputPath = "/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/hello.txt";
            // DataSource也可以换成DataSet
            DataSet<String> stringDataSource = env.readTextFile(inputPath);
            // 对数据集进行处理，按空格分词展开，转换成（word，1）二元组进行统计，
            DataSet<Tuple2<String, Integer>> result = stringDataSource.flatMap(new MyFlatMapper())
                    .groupBy(0) // 按照元组中的第一个位置的word分组
                    .sum(1); // 按照元组中的第二个位置上的数据求和
    
            result.print();
        }
    
        // 自定义类，实现FlatMapFunction接口，java中没有元组的概念，可以使用flink自带的元组类型
        public static class MyFlatMapper implements FlatMapFunction<String, Tuple2<String, Integer>>{ 
    
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception { 
                // 按空格分词
                String[] words = value.split(" ");
                // 遍历所有的word，包成二元组输出
                for(String word : words){ 
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }

**执行结果：**

(demo,2)
    (first,1)
    (flink,4)
    (sunxiaoke,1)
    (do,1)
    (hello,1)
    (it,1)
    (just,1)

### 3.2 流处理单词统计demo ###

StreamWordCount类

package com.test.sun.demo;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @description:
     * @time: 2021-03-21 21:03
     */
    public class StreamWordCount { 
        public static void main(String[] args) throws Exception{ 
            // 创建流处理执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // 从文件中读取数据
            String inputPath = "/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/hello.txt";
            // DataSource也可以换成DataSet
            DataStream<String> stringDataStream = env.readTextFile(inputPath);
    
            // 基于数据流进行转换计算
            DataStream<Tuple2<String, Integer>> resultStream = stringDataStream.flatMap(new WordCount.MyFlatMapper())
                    .keyBy(0)
                    .sum(1);
    
            resultStream.print();
    
            // 执行任务
            env.execute();
        }
    }

**执行结果：**

4> (sunxiaoke,1)
    2> (hello,1)
    2> (demo,1)
    4> (first,1)
    2> (demo,2)
    4> (flink,1)
    1> (it,1)
    4> (flink,2)
    4> (flink,3)
    4> (flink,4)
    2> (just,1)
    2> (do,1)

其中，结果中 > 号之前的数字表示线程号，之所以是1～4是因为我的电脑核数是4。当然，并行的线程数是可以使用setParallelism设置的

增加了并行线程数的代码如下：

package com.test.sun.demo;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @description:
     * @time: 2021-03-21 21:03
     */
    public class StreamWordCount { 
        public static void main(String[] args) throws Exception{ 
            // 创建流处理执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 设置并行执行的线程数
            env.setParallelism(8);
            // 从文件中读取数据
            String inputPath = "/Users/sundongping/IdeaProjects/FirstFlinkTest/src/main/resources/hello.txt";
            // DataSource也可以换成DataSet
            DataStream<String> stringDataStream = env.readTextFile(inputPath);
    
            // 基于数据流进行转换计算
            DataStream<Tuple2<String, Integer>> resultStream = stringDataStream.flatMap(new WordCount.MyFlatMapper())
                    .keyBy(0)
                    .sum(1);
    
            resultStream.print();
    
            // 执行任务
            env.execute();
        }
    }

**执行结果：**

3> (hello,1)
    3> (demo,1)
    7> (sunxiaoke,1)
    3> (just,1)
    3> (do,1)
    1> (it,1)
    7> (flink,1)
    7> (flink,2)
    7> (flink,3)
    7> (first,1)
    7> (flink,4)
    3> (demo,2)

### 3.3 读取socket文本数据demo ###

第一步：  
使用`nc -l -p 7777`命令，作为服务端监听本机的 7777 端口

第二步：  
修改StreamWordCount类代码，将数据源部分改为从socket读取数据

package com.test.sun.demo;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @description:
     * @time: 2021-03-21 21:03
     */
    public class StreamWordCount { 
        public static void main(String[] args) throws Exception{ 
            // 创建流处理执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 设置并行执行的线程数
            env.setParallelism(8);
    
            // 从socket文本流读取数据
            DataStream<String> stringDataStream = env.socketTextStream("localhost", 7777);
    
            // 基于数据流进行转换计算
            DataStream<Tuple2<String, Integer>> resultStream = stringDataStream.flatMap(new WordCount.MyFlatMapper())
                    .keyBy(0)
                    .sum(1);
    
            resultStream.print();
    
            // 执行任务
            env.execute();
        }
    }

第三步：  
在终端输入字符串，编辑器控制台会实时输出统计结果  
![在这里插入图片描述][20210321224922921.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 5]

使用flink提供的ParameterTool工具可以将原来代码中写死的地址和端口号，以参数形式动态传入代码中使用。修改后的代码如下：

package com.test.sun.demo;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.api.java.utils.ParameterTool;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @description:
     * @author: 孙东平
     * @time: 2021-03-21 21:03
     */
    public class StreamWordCount { 
        public static void main(String[] args) throws Exception{ 
            // 创建流处理执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 设置并行执行的线程数
            env.setParallelism(8);
    
            // 用parameter tool工具从程序启动参数中提取配置项
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            String host = parameterTool.get("host");
            Integer port = parameterTool.getInt("port");
    
            // 从socket文本流读取数据
            DataStream<String> stringDataStream = env.socketTextStream(host, port);
    
            // 基于数据流进行转换计算
            DataStream<Tuple2<String, Integer>> resultStream = stringDataStream.flatMap(new WordCount.MyFlatMapper())
                    .keyBy(0)
                    .sum(1);
    
            resultStream.print();
    
            // 执行任务
            env.execute();
        }
    }

然后，配置好启动程序时需要传入的参数即可  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 6]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]: /images/20221022/8d9a38faf2104198a259f0f474f3b8ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]: /images/20221022/419343bc38514c089fbd52d7ba09e456.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210321184828688.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210321184910476.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20210321185036282.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx,size_16,color_FFFFFF,t_70
[20210321224922921.png]: https://img-blog.csdnimg.cn/20210321224922921.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20210321225102474.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20210321230039772.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx,size_16,color_FFFFFF,t_70