大数据技术之Flink---day01概述、快速上手-阿里云开发者社区

Flink是什么

Flink的官网主页地址：https://flink.apache.org/

Flink的核心目标是“数据流上有状态的计算”(Stateful Computations over Data Streams)。

具体说明：Apache Flink是一个“框架和分布式处理引擎”，用于对无界和有界数据流进行有状态计算。

.1.1 无界数据流

有定义流的开始，但是没有定义流的结束
它们会无休止的产生数据
无界流的数据必须持续处理，即数据被摄取后需要立即处理。我们不能等到所有数据都到达再处理，因为输入是无限的。

1.1.2 有界数据流

有定义流的开始，也有定义流的结束
有界流可以在摄取所有数据后再进行计算
有界流所有的数据可以被排序，所有并不需要有序摄取
有界流处理通常被称为批处理

1.1.3 有状态流处理

把流处理需要的额外数据保存成一个“状态”，然后针对这条数据进行处理，并且更新状态，这就是所谓的“有状态的流处理”

状态在内存中：优点：速度快；缺点：可靠性差
状态在分布式系统中：优点：可靠性高；缺点：速度慢

1.1.4 Flink发展历史

1.2 Flink特点

1.3 Flink和SparkStreaming（说实话没有比较的必要）

1、Spark是以批处理为根本。

2、Flink是以流处理为根本。

flink中，每一个算子都可以成为一个独立的task任务

spark中，一个stage是一个taskset(可能包含多个算子)

1.31Flink 的运行时架构

fink和spark一样，都是属于分布式计算引擎

当我们写一个程序，表达我们的计算逻辑，计算本身并不会发生在用户所写的程序中，而是会被一个任务提交系统，把你写的代码中的逻辑，转化成task，然后提交到集群上去分布式并行运行

1.32Flink 的特性

事件驱动型应用:实时风控，实时推荐，实时营销

全量聚合

滚动聚合

为了实现逻辑，在运行中记录的一些数据，在流式计算的语镜中，这些数据叫做状态

如果这个状态数据是用户自己的代码中用变量来记录，一旦系统崩溃，则这个变量中记录的之前状态就丢失，那会让后面的计算产出错误

flink提供了专门的状态管理机

flink可以在系统崩溃后，重启时帮用户自动恢复宕机前的状态

可靠性，容错型，恢复的流程，分布式程序重启后并行度调整后状态的重分配

算子：大逻辑已经确定，而小逻辑有你传入

flink中，每一个算子都可以成为一个独立的task任务

spark中，一个stage是一个taskset(可能包含多个算子)
flink中，每一个算子都可以成为一个独立的task任务，当然，他也可以根据情况，将连续的多个算子，合并成一个task

reparation是作用在rdd上

1.4 Flink的应用场景

1、电商和市场营销

2、物联网（IOT）

3、物流配送和服务业

4、银行和金融业

1.5 Flink分层API

有状态流处理：通过底层API（处理函数），对原始数据加工处理。底层API和DataStreamAPI相集成，可以处理复杂的计算。

DataStreamAPI（流处理）和DataSetAPI（批处理）封装了底层处理函数，提供了通用的模块，比如转换（transformations，包括map，flatMap等），连接（joins），聚合（aggregations），窗口（Windows）操作等。注意：Flink1.12后，DataStreamAPI已经实现真正的流批一体，所以DataSetAPI已经过时。

TableAPI是以表为中心的声明式编程，其中表可能会动态变化。TableAPI遵循关系模型；表有二维数据结构，类似于关系数据库中的表，同时API提供可比较的操作，例如select、project、join、group by、aggregate等。我们可以在表与DataStream/DataSet之间无缝切换，以允许程序将TableAPI与DataStream以及DataSet混用。

SQL这一层在语法与表达能力上与TableAPI类似，但是是以SQL查询表达式的形式表现程序。SQL抽象与TableAPI交互密切，同时SQL查询可以直接在TableAPI定义的表上执行。

2、Flink快速上手

flink 的 DataStream 抽象

2.1 创建项目

在准备好所有的开发环境之后，我们就可以开始开发自己的第一个Flink程序了。首先我们要做的，就是在IDEA中搭建一个Flink项目的骨架。我们会使用Java项目中常见的Maven来进行依赖管理。

1、创建工程

（1）打开IntelliJ IDEA，创建一个Maven工程。

2、添加项目依赖

<properties>
        <flink.version>1.17.0</flink.version>
</properties>


    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>

     <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
     </dependency>
</dependencies>

2.2 WordCount代码编写（大数据常用的例子）

需求：统计一段文字中，每个单词出现的频次

环境准备：创建一个com.zhm.wordcount包

2.2.1 批处理

批处理的基本思路：先逐行读入文件数据，然后将每一行文子拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次。

1、数据准备

（1）在工程根目录下新建一个data文件夹，并在下面创建文本文件words.txt

（2）在文件中输入一些单词

hello hello hello
world world
hello world

2、代码编写

（1）在com.zhm.wordcount包下新建一个Demo01_BatchProcess类

package com.atguigu.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class BatchWordCount {
    public static void main(String[] args) throws Exception {
        //创建支持Flink计算的环境
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        //使用环境去读取数据，封装为计算模型
        DataSource<String> dataSource = executionEnvironment.readTextFile("input/word.txt");
        //调用计算API 3. 转换数据格式 切分、转化(world,1)、分组、聚合
        FlatMapOperator<String, Tuple2<String, Integer>> wordAndOne = dataSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                //3.1按照空格 切分单词
                String[] words = s.split(" ");
                //3.2 将单词转化为(world,1)
                for (String word : words) {
                    Tuple2<String, Integer> wordTuple2 = Tuple2.of(word, 1);
                    //3.3使用Collector 向下游发送数据
                    collector.collect(wordTuple2);
                }
            }
        });
        //4.按照word分组
        UnsortedGrouping<Tuple2<String, Integer>> wordAndOneGroupby = wordAndOne.groupBy(0);
        //5.各分组内聚合
        //1是位置，表示第二个元素
        AggregateOperator<Tuple2<String, Integer>> sum = wordAndOneGroupby.sum(1);
        //6.输出
        sum.print();

    }

}


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
    计算的套路：
        (1) 计算的环境
            Spark：SparkContext
            MR：Driver
            Flink：ExecutionEnvironment
        (2) 把要计算的数据封装为计算模型
            Spark：RDD（Spark Core）
                    DateFrame|DataSet（SparkSQL）
                    DStream（SparkStream）
            MR：k-V
            Flink:DataSource
        (3)调用计算API
            RDD.转换算子()
            MR:自己去编写Mapper、Reducer
            Flink:DataSource.算子()


 */
public class Demo01_BatchProcess {
    public static void main(String[] args) throws Exception {
        //创建支持Flink计算的环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //使用环境去读取数据，封装为计算模型
        DataSource<String> dataSource = env.readTextFile("data/words.txt");
        //调用计算API
        dataSource.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {

            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] split = s.split(" ");
                for (String s1 : split) {
                    collector.collect(new Tuple2<String,Integer>(s1,1));
                }
            }
        }).groupBy(0)
                .sum(1)
                .print();
    }
}

运行结果：

注意：这种实现是基于DataSetAPI的，也就是我们对数据的处理转换，是看作数据集来进行操作的。事实上Flink本身是流批统一的处理架构，批量的数据集本质上也是流，没有必要用两套不同的API来实现。所以从Flink1.12开始，官方推荐的做法是直接使用DataStreamAPI，在提交任务时通过将执行模式设为BATCH来进行批处理；

bin/flink run -Dexecution.runtime-mode=BATCH BatchWordCount.jar

这样，DataSetAPI就没有用了，在实际应用中我们只要维护一套DataStreamAPI就可以。这里只是为了方便大家理解，我们依然用DataSetAPI做了批处理的实现。

2.2.2 流处理

对于Flink而言，流才是整个处理逻辑的底层核心，所以流批一体之后的DataStreamAPI更加强大，可以直接处理批处理和流处理的所有场景。

下面我们就针对不同类型的的输入数据源，用具体的代码来实现流处理。

1、读取文件（有界流）

我们同样试图读取文档words.txt中的数据，并统计每个单词出现的频次。整体思路与之前的批处理非常类似，代码模式也基本一致。

在com.zhm.wordcount包下新建一个Demo02_BoundedStreamProcess类

package com.zhm.wordcount;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


public class Demo02_BoundedStreamProcess {
    public static void main(String[] args) throws Exception {
        //1、创建支持Flink计算的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.1 设置一个线程处理这个流（默认是根据你的cpu数和单词种类个数，取最小值）
//        env.setParallelism(1);

        //2、获取数据源
        FileSource<String> source = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("data/words.txt")).build();
        //3、利用环境将数据源的数据封装为计算模型
        DataStreamSource<String> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "myfile");
        //4、调用API对数据进行计算

        //4.1 将每行数据按照给定的分割符拆分为Tuple2类型的数据模型（word,1）
        streamSource.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] split = s.split(" ");
                for (String s1 : split) {
                    collector.collect(new Tuple2<>(s1,1));
                }
            }
            //4.2 根据word分组
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return stringIntegerTuple2.f0;
            }
            //4.3 根据分组之后，按照元组中的第二列聚相加
        }).sum(1)
                // 4.4 打印结果
                .print();

        //5、提交job
        env.execute();
    }
}

import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
/*
* 接口 A ,里面有一个方法a()
* 正常写法,定义一个class B 实现接口A 方法a()
* B b=new B()
* 匿名实现类:  new A(){
*   a(){
* }
* }
* */

public class StreamWordCount {
    public static void main(String[] args) throws Exception {

        // 1. 创建流式执行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2. 读取文件
        DataStreamSource<String> lineDS = executionEnvironment.readTextFile("input/word.txt");
        // 3. 转换、分组、求和，得到统计结果
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                //按照空格切分
                String[] split = s.split(" ");
                for (String word : split) {
                    //转化成二元组(word,1)
                    Tuple2<String, Integer> wordsAndOne = Tuple2.of(word, 1);
                    //通过 采集器 向下游发送数据
                    collector.collect(wordsAndOne);
                }
            }
        });
        //3.2分组
        KeyedStream<Tuple2<String, Integer>, String> wordAndOneKS = wordAndOneDS.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                return stringIntegerTuple2.f0;
            }
        });
        //3.3聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumDS = wordAndOneKS.sum(1);
        //4.输出数据
        sumDS.print();
        //5.执行:类似 sparkstreming 最后 ssc.start()
        executionEnvironment.execute();
    }
}

运行结果：

和批处理程序BatchWordCount的不同：

创建执行环境的不同，流处理程序使用的是StreamExecutionEnvironment。

转换处理之后，得到的数据对象类型不同

分组操作调用的方法是keyBy方法，可以传入一个匿名函数作为键选择器（KeySelector），指定当前分组的key是什么。

代码末尾需要调用env的execute方法，开始执行任务。

2、读取Socket文本流（无界流）

在实际的生产环境中，真正的数据流其实是无界的，有开始却没有结束，这就要求我们需要持续的处理捕获的数据。为了模拟这种场景，可以监听Socket端口，然后向该端口不断地发生数据。

（1）将StreamWordCount代码中读取文件数据的readTextFile方法，替换成读取Socket文本流的方法socketTextStream。具体代码实现如下:

package com.zhm.wordcount;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


public class Demo03_UnBoundedStreamProcess {
    public static void main(String[] args) throws Exception {

        //1、创建支持Flink计算的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.1 设置一个线程处理这个流
        env.setParallelism(1);

        //2、获取数据源
        DataStreamSource<String> streamSource = env.socketTextStream("hadoop102", 9999);

        //3.1 将每行数据按照给定的分割符拆分为Tuple2类型的数据模型（word,1）
        streamSource.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
                    @Override
                    public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                        String[] split = s.split(" ");
                        for (String s1 : split) {
                            collector.collect(new Tuple2<>(s1,1));
                        }
                    }
                    //3.2 根据word分组
                }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        return stringIntegerTuple2.f0;
                    }
                    //3.3 根据分组之后，按照元组中的第二列聚相加
                }).sum(1)
                // 3.4 打印结果
                .print();

        //4、提交job
        env.execute();
    }
}

（2）在Linux环境的主机hadoop102上，执行下列命令，发送数据进行测试（前提是要安装netcat）

nc -lk hadoop102 9999
• 1

（3）启动Demo03_UnBoundedStreamProcess程序

我们会发现程序启动之后没有任何输出、也不会退出。这是正常的，因为Flink的流处理是事件驱动的，当前程序会一直处于监听状态，只有接受数据才会执行任务、输出统计结果。

（4）从hadoop102发送数据

（5）观察idea控制台

说明：Flink还具有一个类型提前系统，可以分析函数的输入和返回类型，自动获取类型信息

，从而获得对应的序列化器和反序列化器。但是，由于java中泛型擦除的存在，在某些特殊情况下（比如Lambda表达式中），自动提取的信息是不够精细的–只告诉Flink当前的元素由“船头、船身、船尾”构成，根本无法重建出“大船”的模样；这时就需要显示地提供类型信息，才能使得应用程序正常工作或提高其性能。

因为对于flatMap里传入的Lambda表达式，系统只能推断出返回的是Tuple2类型，而无法得到Tuple。只有显示地告诉系统当前的返回类型，才能正确的解析出完整数据。

2.2.3 执行模式

从Flink 1.12开始，官方推荐的做法是直接使用DataStream API，在提交任务时通过将执行模式设为BATCH来进行批处理。不建议使用DataSet API。

// 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
• 1
• 2

DataStreamAPI执行模式包括：流执行模式、批执行模式和自动模式。

流执行模式（Streaming）

这是DataStreamAPI最经典的模式，一边用于需要持续实时处理的无界数据流。默认情况下，程序使用的就是Streaming执行模式。

批执行模式（Batch）

专门用于批处理的执行模式

自动模式

在这种模式下，将由程序根据输入数据源是否有界来自动选择执行模式。

批执行模式的使用：主要有两种方式：

（1）通过命令行配置

bin/flink run -Dexecution.runtime-mode=BATCH ...
• 1

在提交作业时，增加execution.runtime-mode参数，指定值为BATCH。

（2）通过代码设置

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

在代码中，直接基于执行环境调用setRuntimeMode方法，传入BATCH模式。

实际应用中一般不会在代码中配置，而是使用命令行，这样更加灵活。

2.2.4 本地WebUI

在Idea本地运行程序时，可以通过添加本地WebUI依赖，使用WebUI界面查看Job的运行情况。

  <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

添加后，在代码中可以指定绑定的端口:

Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

大数据技术之Flink---day01概述、快速上手

Flink是什么

.1.1 无界数据流

1.1.2 有界数据流

1.1.3 有状态流处理

1.1.4 Flink发展历史

1.2 Flink特点

1.3 Flink和SparkStreaming（说实话没有比较的必要）

1.31Flink 的运行时架构

1.32Flink 的特性

1.4 Flink的应用场景

1.5 Flink分层API

2、Flink快速上手

flink 的 DataStream 抽象

2.1 创建项目

2.2 WordCount代码编写（大数据常用的例子）

2.2.1 批处理

2.2.2 流处理

2.2.3 执行模式

2.2.4 本地WebUI

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

大数据技术之Flink---day01概述、快速上手

Flink是什么

.1.1 无界数据流

1.1.2 有界数据流

1.1.3 有状态流处理

1.1.4 Flink发展历史

1.2 Flink特点

1.3 Flink和SparkStreaming（说实话没有比较的必要）

1.31Flink 的运行时架构

1.32Flink 的特性

1.4 Flink的应用场景

1.5 Flink分层API

2、Flink快速上手

flink 的 DataStream 抽象

2.1 创建项目

2.2 WordCount代码编写（大数据常用的例子）

2.2.1 批处理

2.2.2 流处理

2.2.3 执行模式

2.2.4 本地WebUI

热门文章

最新文章

相关课程

相关电子书

相关实验场景