1 Flink简介

Apache Flink® — Stateful Computations over Data Streams

Apache Flink 是一个框架和分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。Flink 能在所有常见集群环境中运行，并能以内存速度和任意规模进行计算。

1.1 处理无界和有界数据

任何类型的数据都可以形成一种事件流。信用卡交易、传感器测量、机器日志、网站或移动应用程序上的用户交互记录，所有这些数据都形成一种流。

数据可以被作为 无界 或者 有界 流来处理。

无界流 有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。

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

Apache Flink 擅长处理无界和有界数据集 精确的时间控制和状态化使得 Flink 的运行时(runtime)能够运行任何处理无界流的应用。有界流则由一些专为固定大小数据集特殊设计的算法和数据结构进行内部处理，产生了出色的性能。

1.2 部署应用到任意地方

Apache Flink 是一个分布式系统，它需要计算资源来执行应用程序。Flink 集成了所有常见的集群资源管理器，例如 Hadoop YARN、 Apache Mesos 和 Kubernetes，但同时也可以作为独立集群运行。 Flink 被设计为能够很好地工作在上述每个资源管理器中，这是通过资源管理器特定(resource-manager-specific)的部署模式实现的。Flink 可以采用与当前资源管理器相适应的方式进行交互。 部署 Flink 应用程序时，Flink 会根据应用程序配置的并行性自动标识所需的资源，并从资源管理器请求这些资源。在发生故障的情况下，Flink 通过请求新资源来替换发生故障的容器。提交或控制应用程序的所有通信都是通过 REST 调用进行的，这可以简化 Flink 与各种环境中的集成

1.3 运行任意规模应用

Flink 旨在任意规模上运行有状态流式应用。因此，应用程序被并行化为可能数千个任务，这些任务分布在集群中并发执行。所以应用程序能够充分利用无尽的 CPU、内存、磁盘和网络 IO。而且 Flink 很容易维护非常大的应用程序状态。其异步和增量的检查点算法对处理延迟产生最小的影响，同时保证精确一次状态的一致性。 Flink 用户报告了其生产环境中一些令人印象深刻的扩展性数字： 每天处理数万亿的事件 可以维护几TB大小的状态 可以部署上千个节点的集群

1.4 利用内存性能

有状态的 Flink 程序针对本地状态访问进行了优化。任务的状态始终保留在内存中，如果状态大小超过可用内存，则会保存在能高效访问的磁盘数据结构中。任务通过访问本地（通常在内存中）状态来进行所有的计算，从而产生非常低的处理延迟。Flink 通过定期和异步地对本地状态进行持久化存储来保证故障场景下精确一次的状态一致性。

2 Flink架构图

3 入门案例演示

3.1 实时需求分析

实时统计每隔1秒统计最近2秒单词出现的次数

3.2 开发环境部署

官网建议使用IDEA，IDEA默认集成了scala和maven，使用起来方便

本次课使用了当前最新的flink版本

pom依赖如下：

<properties> <flink.version>1.11.1</flink.version> <scala.version>2.12.12</scala.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> <version>${flink.version}</version> </dependency> </dependencies>

3.3 实时代码开发（java）

import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.util.Collector; public class WindowWordCountJava { public static void main(String[] args) throws Exception { //flink提供的工具类，获取传递的参数 ParameterTool parameterTool = ParameterTool.fromArgs(args); String hostname = parameterTool.get("hostname"); int port = parameterTool.getInt("port"); //步骤一：获取执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //步骤二：获取数据源 DataStream<String> dataStream = env.socketTextStream(hostname, port); //步骤三：执行逻辑操作 DataStream<WordCount> wordAndOneStream = dataStream.flatMap(new FlatMapFunction<String, WordCount>() { public void flatMap(String line, Collector<WordCount> out) { String[] fields = line.split(","); for (String word : fields) { out.collect(new WordCount(word, 1L)); } } }); DataStream<WordCount> resultStream = wordAndOneStream.keyBy("word") .timeWindow(Time.seconds(2), Time.seconds(1))//每隔1秒计算最近2秒 .sum("count"); //步骤四：结果打印 resultStream.print(); //步骤五：任务启动 env.execute("WindowWordCountJava"); } public static class WordCount{ public String word; public long count; //记得要有这个空构建 public WordCount(){ } public WordCount(String word,long count){ this.word = word; this.count = count; } @Override public String toString() { return "WordCount{" + "word='" + word + '\'' + ", count=" + count + '}'; } } }

3.4 实时代码开发（scala）

添加依赖

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

scala开发依赖和编译插件

<properties> <flink.version>1.11.1</flink.version> <scala.version>2.12.12</scala.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.11</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_2.11</artifactId> <version>${flink.version}</version> </dependency> </dependencies> <build> <pluginManagement> <plugins> <!-- scala插件 --> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.2.2</version> </plugin> <!-- maven 插件 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.5.1</version> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> </plugins> </pluginManagement> <plugins> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <executions> <execution> <id>scala-compile-first</id> <phase>process-resources</phase> <goals> <goal>add-source</goal> <goal>compile</goal> </goals> </execution> <execution> <id>scala-test-compile</id> <phase>process-test-resources</phase> <goals> <goal>testCompile</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <executions> <execution> <phase>compile</phase> <goals> <goal>compile</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>2.4.3</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> </configuration> </execution> </executions> </plugin> </plugins> </build>

代码开发

import org.apache.flink.api.java.utils.ParameterTool import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.streaming.api.windowing.time.Time /** * 滑动窗口 * 每隔1秒钟统计最近2秒内的数据，打印到控制台。 */ object WindowWordCountScala { def main(args: Array[String]): Unit = { //获取参数 val hostname = ParameterTool.fromArgs(args).get("hostname") val port = ParameterTool.fromArgs(args).getInt("port") //TODO 导入隐式转换 import org.apache.flink.api.scala._ //步骤一：获取执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //步骤二：获取数据源 val textStream = env.socketTextStream(hostname,port) //步骤三：数据处理 val wordCountStream = textStream.flatMap(line => line.split(",")) .map((_, 1)) .keyBy(0) .timeWindow(Time.seconds(2), Time.seconds(1)) .sum(1) //步骤四：数据结果处理 wordCountStream.print() //步骤六：启动程序 env.execute("WindowWordCountScala") } }

3.5 需求分析

对文件进行单词计数

3.6 离线代码开发（java）

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.tuple.Tuple2; import org.apache.flink.util.Collector; public class WordCount { public static void main(String[] args)throws Exception { //步骤一：获取离线的程序入口 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); String inputPath="D:\\kkb\\flinklesson\\src\\main\\input\\hello.txt"; //步骤二：获取数据源 DataSource<String> dataSet = env.readTextFile(inputPath); //步骤三：数据处理 FlatMapOperator<String, Tuple2<String, Integer>> wordAndOneDataSet = dataSet.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() { @Override public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception { String[] fileds = line.split(","); for (String word : fileds) { collector.collect(new Tuple2<String, Integer>(word, 1)); } } }); AggregateOperator<Tuple2<String, Integer>> result = wordAndOneDataSet.groupBy(0) .sum(1); //步骤四：数据结果处理 result.writeAsText("D:\\kkb\\flinklesson\\src\\output\\result").setParallelism(1); //步骤五：启动程序 env.execute("word count"); } }

换一种写法

public class WordCount { public static void main(String[] args)throws Exception { //步骤一：获取离线的程序入口 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); String inputPath="D:\\kkb\\flinklesson\\src\\main\\input\\hello.txt"; //步骤二：获取数据源 DataSource<String> dataSet = env.readTextFile(inputPath); //步骤三：数据处理 FlatMapOperator<String, Tuple2<String, Integer>> wordAndOneDataSet = dataSet.flatMap(new MySplitWordsTask()); AggregateOperator<Tuple2<String, Integer>> result = wordAndOneDataSet.groupBy(0) .sum(1); //步骤四：数据结果处理 result.writeAsText("D:\\kkb\\flinklesson\\src\\output\\result1").setParallelism(1); //步骤五：启动程序 env.execute("word count"); } public static class MySplitWordsTask implements FlatMapFunction<String,Tuple2<String,Integer>>{ @Override public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception { String[] fileds = line.split(","); for (String word : fileds) { collector.collect(new Tuple2<String, Integer>(word, 1)); } } } }

总结：上面这种写法，我们把flink的算子抽离出来，代码看起来会更清晰。

4 local模式安装

4.1Local模式安装

（1）安装jdk，配置JAVA_HOME，建议使用jdk1.8以上

（2）安装包下载地址：http://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.11.1/flink-1.11.1-bin-scala_2.12.tgz

（3）直接上传安装包到服务器

​ 解压安装包：tar -zxvf flink-1.11.1-bin-scala_2.12.tgz

​ 创建快捷方式： ln -s flink-1.11.1-bin-scala_2.12.tgz flink

​ 配置FLINK_HOEM环境变量

（4）启动服务

​ local模式，什么配置项都不需要配，直接启动服务器即可

​ cd $FLIKE_HOME

​ ./bin/start-cluster.sh 启动服务

​ ./bin/stop-cluster.sh 停止服务

（5）Web页面浏览

​ localhost:8081

4.2 在集群上提交任务

（1）编译，打包

​ mvn clean package

​ pom文件里面的依赖上添加上：

​ provided

​ 代表这个依赖打包的时候不打到jar包里面，因为我们这些依赖包flink已经自带了，所以我们打包的时候不用把打包进去。

(2) 运行

​ 在hadoop01 上执行 nc -lk 9999

​ flink run -c stream.lesson01.WindowWordCountJava flinklesson-1.0-SNAPSHOT.jar -port 9999

查看日志：

tail -F $FLINK_HOME/log/flink-work-taskexecutor-0-tjtx-98-197.58os.org.out

(3) 停止任务

flink cancel job-id

5 Standalone模式安装

**（1）集群规划 **

主机名JobManagerTaskManagerhadoop01是hadoop02是hadoop03是

**（2）依赖 **

jdk1.8以上，配置JAVA_HOME

主机之间免密码

flink-1.11.1-bin-scala_2.12.tgz

**（3）安装步骤 **

(a) 修改conf/flink-conf.yaml jobmanager.rpc.address: hadoop01 (b) 修改conf/slaves hadoop02 hadoop03 (c) 拷贝到其他节点 scp -rq /usr/local/flink-1.11.1 hadoop02:/usr/local scp -rq /usr/local/flink-1.11.1 hadoop03:/usr/local (d)：在hadoop01(JobMananger)节点启动 start-cluster.sh (e)：访问http://hadoop01:8081

**(4) StandAlone模式需要考虑的参数 **

jobmanager.heap.mb：jobmanager节点可用的内存大小

taskmanager.heap.mb：taskmanager节点可用的内存大小

taskmanager.numberOfTaskSlots：每台机器可用的cpu数量

parallelism.default：默认情况下任务的并行度

taskmanager.tmp.dirs：taskmanager的临时数据存储目录

6 Flink on Yarn模式安装

flink on yarn有两种方式：

4.6.1 第一种方式

在YARN里面启动一个flink集群，然后我们再往flink集群提交任务，除非把Flink集群停了，不然资源不会释放

6.2 第二种方式

没提交一个任务就在yarn上面启动一个flink小集群（推荐使用）

任务运行完了资源就自动释放

6.3 不同模式的任务提交

第一种【yarn-session.sh(开辟资源)+flink run(提交任务)】 启动一个一直运行的flink集群 /bin/yarn-session.sh -n 2 -jm 1024 -tm 1024 [-d] 把任务附着到一个已存在的flink yarn session •./bin/yarn-session.sh -id application_1463870264508_0029 •执行任务 •./bin/flink run ./examples/batch/WordCount.jar -input hdfs://hadoop100:9000/LICENSE -output hdfs://hadoop100:9000/wordcount-result.txt 停止任务 【web界面或者命令行执行cancel命令】 第二种【flink run -m yarn-cluster(开辟资源+提交任务)】 •启动集群，执行任务 •./bin/flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar 注意：client端必须要设置YARN_CONF_DIR或者HADOOP_CONF_DIR或者HADOOP_HOME环境变量，通过这个环境变量来读取YARN和HDFS的配置信息，否则启动会失败

help信息

yarn-session.sh

用法: 必选 -n,--container <arg> 分配多少个yarn容器 (=taskmanager的数量) 可选 -D <arg> 动态属性 -d,--detached 独立运行 -jm,--jobManagerMemory <arg> JobManager的内存 [in MB] -nm,--name 在YARN上为一个自定义的应用设置一个名字 -q,--query 显示yarn中可用的资源 (内存, cpu核数) -qu,--queue <arg> 指定YARN队列. -s,--slots <arg> 每个TaskManager使用的slots数量 -tm,--taskManagerMemory <arg> 每个TaskManager的内存 [in MB] -z,--zookeeperNamespace <arg> 针对HA模式在zookeeper上创建NameSpace -id,--applicationId <yarnAppId> YARN集群上的任务id，附着到一个后台运行的yarn session中

flink run

run [OPTIONS] <jar-file> <arguments> "run" 操作参数: -c,--class <classname> 如果没有在jar包中指定入口类，则需要在这里通过这个参数指定 -m,--jobmanager <host:port> 指定需要连接的jobmanager(主节点)地址，使用这个参数可以指定一个不同于配置文件中的jobmanager -p,--parallelism <parallelism> 指定程序的并行度。可以覆盖配置文件中的默认值。 默认查找当前yarn集群中已有的yarn-session信息中的jobmanager【/tmp/.yarn-properties-root】： ./bin/flink run ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1 连接指定host和port的jobmanager： ./bin/flink run -m hadoop100:1234 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1 启动一个新的yarn-session： ./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1 注意：yarn session命令行的选项也可以使用./bin/flink 工具获得。它们都有一个y或者yarn的前缀 例如：./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar

6.4 Flink on YARN集群部署

(1) flink on yarn运行原理

其实Flink on YARN部署很简单，就是只要部署好hadoop集群即可，我们只需要部署一个Flink客户端，然后从flink客户端提交Flink任务即可。

7 Flink Shell使用

针对初学者，开发的时候容易出错，如果每次都打包进行调试，比较麻烦，并且也不好定位问题，可以在scala shell命令行下进行调试

scala shell方式支持流处理和批处理。当启动shell命令行之后，两个不同的ExecutionEnvironments会被自动创建。使用senv(Stream)和benv(Batch)分别去处理流处理和批处理程序。(类似于spark-shell中sc变量)

bin/start-scala-shell.sh [local|remote|yarn] [options]

8 Flink之数据源

8.1 source简介

source是程序的数据源输入，你可以通过StreamExecutionEnvironment.addSource(sourceFunction)来为你的程序添加一个source。

flink提供了大量的已经实现好的source方法，你也可以自定义source：

（1）通过实现sourceFunction接口来自定义无并行度的source

（2）通过实现ParallelSourceFunction 接口 or 继承RichParallelSourceFunction 来自定义有并行度的source

不过大多数情况下，我们使用自带的source即可。

获取source的方式

（1）基于文件

​ readTextFile(path)

​ 读取文本文件，文件遵循TextInputFormat 读取规则，逐行读取并返回。

（2）基于socket

​ socketTextStream ​ 从socker中读取数据，元素可以通过一个分隔符切开。

（3）基于集合

​ fromCollection(Collection)

​ 通过java 的collection集合创建一个数据流，集合中的所有元素必须是相同类型的。

（4）自定义输入

​ addSource 可以实现读取第三方数据源的数据

​ 系统内置提供了一批connectors，连接器会提供对应的source支持【kafka】

自带的connectors

Apache Kafka (source/sink) 后面重点分析Apache Cassandra (sink)Amazon Kinesis Streams (source/sink)Elasticsearch (sink)Hadoop FileSystem (sink)RabbitMQ (source/sink)Apache NiFi (source/sink)Twitter Streaming API (source)

8.2 数据源之collection

public class StreamingSourceFromCollection { public static void main(String[] args) throws Exception { //步骤一：获取环境变量 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //步骤二：模拟数据 ArrayList<String> data = new ArrayList<String>(); data.add("hadoop"); data.add("spark"); data.add("flink"); //步骤三：获取数据源 DataStreamSource<String> dataStream = env.fromCollection(data); //步骤四：transformation操作 SingleOutputStreamOperator<String> addPreStream = dataStream.map(new MapFunction<String, String>() { @Override public String map(String word) throws Exception { return "kaikeba_" + word; } }); //步骤五：对结果进行处理（打印） addPreStream.print().setParallelism(1); //步骤六：启动程序 env.execute("StreamingSourceFromCollection"); } }

8.3 自定义单并行度数据源

/** * 注意：指定数据类型 * 功能：每秒产生一条数据 */ public class MyNoParalleSource implements SourceFunction<Long> { private long number = 1L; private boolean isRunning = true; @Override public void run(SourceContext<Long> sct) throws Exception { while (isRunning){ sct.collect(number); number++; //每秒生成一条数据 Thread.sleep(1000); } } @Override public void cancel() { isRunning=false; } } /** * 功能：从自定义的数据数据源里面获取数据，然后过滤出偶数 */ public class StreamingDemoWithMyNoPralalleSource { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<Long> numberStream = env.addSource(new MyNoParalleSource()).setParallelism(1); SingleOutputStreamOperator<Long> dataStream = numberStream.map(new MapFunction<Long, Long>() { @Override public Long map(Long value) throws Exception { System.out.println("接受到了数据："+value); return value; } }); SingleOutputStreamOperator<Long> filterDataStream = dataStream.filter(new FilterFunction<Long>() { @Override public boolean filter(Long number) throws Exception { return number % 2 == 0; } }); filterDataStream.print().setParallelism(1); env.execute("StreamingDemoWithMyNoPralalleSource"); } }

运行结果：

接受到了数据：1 接受到了数据：2 2 接受到了数据：3 接受到了数据：4 4 接受到了数据：5 接受到了数据：6 6 接受到了数据：7 接受到了数据：8 8

8.4 自定义多并行度数据源

/** * 功能：自定义支持并行度的数据源 * 每秒产生一条数据 */ public class MyParalleSource implements ParallelSourceFunction<Long> { private long number = 1L; private boolean isRunning = true; @Override public void run(SourceContext<Long> sct) throws Exception { while (isRunning){ sct.collect(number); number++; //每秒生成一条数据 Thread.sleep(1000); } } @Override public void cancel() { isRunning=false; } } public class StreamingDemoWithMyPralalleSource { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<Long> numberStream = env.addSource(new MyParalleSource()).setParallelism(2); SingleOutputStreamOperator<Long> dataStream = numberStream.map(new MapFunction<Long, Long>() { @Override public Long map(Long value) throws Exception { System.out.println("接受到了数据："+value); return value; } }); SingleOutputStreamOperator<Long> filterDataStream = dataStream.filter(new FilterFunction<Long>() { @Override public boolean filter(Long number) throws Exception { return number % 2 == 0; } }); filterDataStream.print().setParallelism(1); env.execute("StreamingDemoWithMyNoPralalleSource"); } }

运行结果：

接受到了数据：1 接受到了数据：1 接受到了数据：2 接受到了数据：2 2 2 接受到了数据：3 接受到了数据：3 接受到了数据：4 4 接受到了数据：4 4 接受到了数据：5 接受到了数据：5 接受到了数据：6 接受到了数据：6