FlinK 1.12批流一体
flink 1.13.0 流处理应用更加简单高效
首先第一代的计算引擎,无疑就是 Hadoop 承载的 MapReduce。它将计算分为两个阶段,分别为 Map 和 Reduce。对于上层应用来说,就不得不想方设法去拆分算法,甚至于不得不在上层应用实现多个 Job 的串联,以完成一个完整的算法,例如迭代计算。
n 批处理
n Mapper、Reducer
由于这样的弊端,催生了支持 DAG 框架的产生。因此,支持 DAG 的框架被划分为第二代计算引擎。如 Tez 以及更上层的 Oozie。这里我们不去细究各种 DAG 实现之间的区别,不过对于当时的 Tez 和 Oozie 来说,大多还是批处理的任务。
n 批处理
n 1个Tez = MR(1) + MR(2) + ... + MR(n)
n 相比MR效率有所提升
接下来就是以 Spark 为代表的第三代的计算引擎。第三代计算引擎的特点主要是 Job 内部的 DAG 支持(不跨越 Job),以及强调的实时计算。在这里,很多人也会认为第三代计算引擎也能够很好的运行批处理的 Job。
n 批处理、流处理、SQL高层API支持
n 自带DAG
内存迭代计算、性能较之前大幅提升
随着第三代计算引擎的出现,促进了上层应用快速发展,例如各种迭代计算的性能以及对流计算和 SQ- 等的支持。Flink 的诞生就被归在了第四代。这应该主要表现在 Flink 对流计算的支持,以及更一步的实时性上面。当然 Flink 也可以支持 Batch 的任务,以及 DAG 的运算。
n 批处理、流处理、SQL高层API支持
n 自带DAG
n 流式计算性能更高、可靠性更高
事件驱动
数据分析
数据管道
也称之为Master,用于协调分布式执行,它们用来调度task,协调检查点,协调失败时恢复等。Flink运行时至少存在一个master处理器,如果配置高可用模式则会存在多个master处理器,它们其中有一个是leader,而其他的都是standby。
也称之为Worker,用于执行一个dataflow的task(或者特殊的subtask)、数据缓冲和data stream的交换,Flink运行时至少会存在一个worker处理器。
物理概念,一个TM(TaskManager)内会划分出多个Slot,1个Slot内最多可以运行1个Task(Subtask)或一组由Task(Subtask)组成的任务链。
多个Slot之间会共享平分当前TM的内存空间
Slot是对一个TM的资源进行固定分配的工具,每个Slot在TM启动后,可以获得固定的资源
比如1个TM是一个JVM进程,如果有6个Slot,那么这6个Slot平分这一个JVM进程的资源
但是因为在同一个进程内,所以线程之间共享TCP连接、内存数据等,效率更高(Slot之间交流方便)
任务,每一个Flink的Job会根据情况(并行度、算子类型)将一个整体的Job划分为多个Task
子任务,一个Task可以由多个Subtask组成,一个Task有多少个Subtask取决于这个Task的并行度
也就是,每一个Subtask就是当前Task任务并行的一个线程
如,当前Task并行度为8,那么这个Task会有8个Subtask(8个线程并行执行这个Task)
并行度就是一个Task可以分成多少个Subtask并行执行的一个参数
这个参数是动态的,可以在任务执行前进行分配,而非Slot分配,TM启动就固定了
一个Task可以获得的最大并行度取决于整个Flink环境的可用Slot数量,也就是如果有8个Slot,那么最大并行度也就是8,设置的再大也没有意义
如下图:
- 一个Job分为了3个Task来运行,分别是TaskA TaskB TaskC
- 其中TaskA设置为了6个并行度,也就是TaskA可以有6个Subtask,如图可见,TaskA的6个Subtask各自在一个Slot内执行
- 其中在Slot的时候说过,Slot可以运行由Task(或Subtask)组成的任务链,如图可见,最左边的Slot运行了TaskA TaskB TaskC 3个Task各自的1个Subtask组成的一个Subtask执行链
Slot是物理的概念,是静态的概念,一旦flink启动以后,tm就制定了slot数量,不能改变
parallelism是动态的概念,可以设置并行度的优先级,可以设置算子级别的或者应用程序全局的并行度、递交作业时设置并行度、使用flink部署环境默认配置文件中指定的并行度
优先级从前往后,越来越低
并行度是一个动态的概念,可以在多个地方设置并行度:
- 配置文件默认并行度:conf/flink-conf.yaml的parallelism.default
- 启动Flink任务,动态提交参数:比如:bin/flink run -p 3
- 在代码中设置全局并行度:env.setParallelism(3);
- 针对每个算子进行单独设置:sum(1).setParallelism(3)
优先级:算子 > 代码全局 > 命令行参数 > 配置文件
最顶层:SQL/Table API 提供了操作关系表、执行SQL语句分析的API库,供我们方便的开发SQL相关程序
中层:流和批处理API层,提供了一系列流和批处理的API和算子供我们对数据进行处理和分析
最底层:运行时层,提供了对Flink底层关键技术的操纵,如对Event、state、time、window等进行精细化控制的操作API
支持机器学习(FlinkML)
支持图分析(Gelly)
支持关系数据处理(Table)
支持复杂事件处理(CEP) 风控领域用得特别多
maven
<?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>cn.itcast</groupId> <artifactId>itcast_flinkbase51</artifactId> <version>1.0-SNAPSHOT</version> <!-- 指定仓库位置,依次为aliyun、apache和cloudera仓库 --> <repositories> <!-- <repository>--> <!-- <id>aliyun</id>--> <!-- <url>http://maven.aliyun.com/nexus/content/groups/public/</url>--> <!-- </repository>--> <repository> <id>apache</id> <url>https://repository.apache.org/content/repositories/snapshots/</url> </repository> <repository> <id>cloudera</id> <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url> </repository> </repositories> <properties> <encoding>UTF-8</encoding> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <java.version>1.8</java.version> <scala.version>2.11</scala.version> <flink.version>1.13.1</flink.version> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> </properties> <dependencies> <!--依赖Scala语言--> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>2.12.11</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-scala_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- web ui的依赖 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-runtime-web_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- Apache Flink 的依赖 --> <!-- 这些依赖项,不应该打包到JAR文件中. --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_2.12</artifactId> <version>${flink.version}</version> </dependency> <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-table-api-scala-bridge_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge_2.12</artifactId> <version>${flink.version}</version> </dependency> <!-- blink执行计划,1.11+默认的--> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-blink_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>${flink.version}</version> </dependency> <!--<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-cep_2.12</artifactId> <version>${flink.version}</version> </dependency>--> <!-- flink连接器--> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-sql-connector-kafka_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-jdbc_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-csv</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-json</artifactId> <version>${flink.version}</version> </dependency> <!-- <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-filesystem_2.12</artifactId> <version>${flink.version}</version> </dependency>--> <!--<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-jdbc_2.12</artifactId> <version>${flink.version}</version> </dependency>--> <!--<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-parquet_2.12</artifactId> <version>${flink.version}</version> </dependency>--> <!--<dependency> <groupId>org.apache.avro</groupId> <artifactId>avro</artifactId> <version>1.9.2</version> </dependency> <dependency> <groupId>org.apache.parquet</groupId> <artifactId>parquet-avro</artifactId> <version>1.10.0</version> </dependency>--> <dependency> <groupId>org.apache.bahir</groupId> <artifactId>flink-connector-redis_2.11</artifactId> <version>1.0</version> <exclusions> <exclusion> <artifactId>flink-streaming-java_2.11</artifactId> <groupId>org.apache.flink</groupId> </exclusion> <exclusion> <artifactId>flink-runtime_2.11</artifactId> <groupId>org.apache.flink</groupId> </exclusion> <exclusion> <artifactId>flink-core</artifactId> <groupId>org.apache.flink</groupId> </exclusion> <exclusion> <artifactId>flink-java</artifactId> <groupId>org.apache.flink</groupId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-hive_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.hive</groupId> <artifactId>hive-metastore</artifactId> <version>2.1.0</version> <exclusions> <exclusion> <artifactId>hadoop-hdfs</artifactId> <groupId>org.apache.hadoop</groupId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.hive</groupId> <artifactId>hive-exec</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-shaded-hadoop-2-uber</artifactId> <version>2.7.5-10.0</version> </dependency> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.38</version> <!--<version>8.0.20</version>--> </dependency> <!-- 高性能异步组件:Vertx--> <dependency> <groupId>io.vertx</groupId> <artifactId>vertx-core</artifactId> <version>3.9.0</version> </dependency> <dependency> <groupId>io.vertx</groupId> <artifactId>vertx-jdbc-client</artifactId> <version>3.9.0</version> </dependency> <dependency> <groupId>io.vertx</groupId> <artifactId>vertx-redis-client</artifactId> <version>3.9.0</version> </dependency> <!-- 日志 --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.7</version> <scope>runtime</scope> </dependency> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</version> <scope>runtime</scope> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.44</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.2</version> <scope>provided</scope> </dependency> <!-- 参考:https://blog.csdn.net/f641385712/article/details/84109098--> <!--<dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-collections4</artifactId> <version>4.4</version> </dependency>--> <!--<dependency> <groupId>org.apache.thrift</groupId> <artifactId>libfb303</artifactId> <version>0.9.3</version> <type>pom</type> <scope>provided</scope> </dependency>--> <!--<dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>28.2-jre</version> </dependency>--> </dependencies> <build> <sourceDirectory>src/main/java</sourceDirectory> <plugins> <!-- 编译插件 --> <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> <!--<encoding>${project.build.sourceEncoding}</encoding>--> </configuration> </plugin> <!-- 打包插件(会包含所有依赖) --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>2.3</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <!-- zip -d learn_spark.jar META-INF/*.RSA META-INF/*.DSA META-INF/*.SF --> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <!-- 设置jar包的入口类(可选) --> <mainClass></mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build> </project>
package cn.itcast.day01; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.operators.*; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; 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; /** * todo flink1.12之前的写法 * 编写Flink程序,接收socket的单词数据,并以空格进行单词拆分打印。 */ public class StreamWordCount { public static void main(String[] args) throws Exception { /** * 实现步骤: * 1)获取flink流处理的运行环境 * 2)接入数据源,读取文件获取数据 * 3)数据处理 * 3.1:使用flatMap对单词进行拆分 * 3.2:对拆分后的单词进行记一次数 * 3.3:使用分组算子对key进行分组 * 3.4:对分组后的key进行聚合操作 * 4)构建sink,输出结果 */ //todo 1)获取flink流处理的运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //todo 2)接入数据源,读取文件获取数据 DataStreamSource<String> lines = env.socketTextStream("node01", 9999); //todo 3)数据处理 // 3.1:使用flatMap对单词进行拆分 SingleOutputStreamOperator<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String line, Collector<String> out) throws Exception { String[] words = line.split(" "); //返回数据 for (String word : words) { out.collect(word); } } }); // 3.2:对拆分后的单词进行记一次数 SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = words.map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String word) throws Exception { return Tuple2.of(word, 1); } }); // 3.3:使用分组算子对key进行分组 //wordAndOne.keyBy(0); // KeyedStream<Tuple2<String, Integer>, String> grouped = wordAndOne.keyBy(new KeySelector<Tuple2<String, Integer>, String>() { // @Override // public String getKey(Tuple2<String, Integer> value) throws Exception { // return value.f0; // } // }); KeyedStream<Tuple2<String, Integer>, String> grouped = wordAndOne.keyBy(t -> t.f0); // 3.4:对分组后的key进行聚合操作 SingleOutputStreamOperator<Tuple2<String, Integer>> sumed = grouped.sum(1); //todo 4)构建sink,输出结果 sumed.print(); //todo 5)启动运行 env.execute(); } }
package cn.itcast.day01; import groovy.lang.Tuple; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.operators.*; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.util.Collector; /** * todo flink1.12之前的写法 * 编写Flink程序,读取文件中的字符串,并以空格进行单词拆分打印。 */ public class BatchWordCount { public static void main(String[] args) throws Exception { /** * 实现步骤: * 1)获取flink批处理的运行环境 * 2)接入数据源,读取文件获取数据 * 3)数据处理 * 3.1:使用flatMap对单词进行拆分 * 3.2:对拆分后的单词进行记一次数 * 3.3:使用分组算子对key进行分组 * 3.4:对分组后的key进行聚合操作 * 4)构建sink,输出结果 */ //todo 1)获取flink批处理的运行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); //todo 2)接入数据源,读取文件获取数据 DataSource<String> lines = env.readTextFile("./data/input/wordcount.txt"); //todo 3)数据处理 // 3.1:使用flatMap对单词进行拆分 FlatMapOperator<String, String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String line, Collector<String> out) throws Exception { String[] words = line.split(" "); //返回数据 for (String word : words) { out.collect(word); } } }); // 3.2:对拆分后的单词进行记一次数 MapOperator<String, Tuple2<String, Integer>> wordAndOne = words.map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String word) throws Exception { return Tuple2.of(word, 1); } }); // 3.3:使用分组算子对key进行分组 UnsortedGrouping<Tuple2<String, Integer>> grouped = wordAndOne.groupBy(0); // 3.4:对分组后的key进行聚合操作 AggregateOperator<Tuple2<String, Integer>> sumed = grouped.sum(1); //todo 4)构建sink,输出结果 sumed.print(); } }
package cn.itcast.day01; import org.apache.flink.api.common.RuntimeExecutionMode; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; 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; /** * todo flink1.12以后的实现方案 * todo 使用批流一体API编程模型实现单词计数 * * 在flink中批是流的一个特例,也就意味着不管实现批还是流处理,肯定按照流的api实现批处理 * DataStream * StreamExecutionEnvironment */ public class UnifyWordCount { public static void main(String[] args) throws Exception { /** * 实现步骤: * 1)获取flink流处理的运行环境 * 2)接入数据源,读取文件获取数据 * 3)数据处理 * 3.1:使用flatMap对单词进行拆分 * 3.2:对拆分后的单词进行记一次数 * 3.3:使用分组算子对key进行分组 * 3.4:对分组后的key进行聚合操作 * 4)构建sink,输出结果 */ //todo 1)获取flink流处理的运行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); //使用dataStream实现批处理 //env.setRuntimeMode(RuntimeExecutionMode.STREAMING); //使用dataStream实现流处理(如果数据源是一个有界数据流则依然是一个批处理) //env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//使用DataStream的时候根据数据源自动选择是批还是流 //todo 2)接入数据源,读取文件获取数据 //DataStreamSource<String> lines = env.readTextFile("./data/input/wordcount.txt"); DataStreamSource<String> lines = env.socketTextStream("node01", 9999); //todo 3)数据处理 // 3.1:使用flatMap对单词进行拆分 SingleOutputStreamOperator<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String line, Collector<String> out) throws Exception { String[] words = line.split(" "); //返回数据 for (String word : words) { out.collect(word); } } }); // 3.2:对拆分后的单词进行记一次数 SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = words.map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String word) throws Exception { return Tuple2.of(word, 1); } }); // 3.3:使用分组算子对key进行分组 //wordAndOne.keyBy(0); // KeyedStream<Tuple2<String, Integer>, String> grouped = wordAndOne.keyBy(new KeySelector<Tuple2<String, Integer>, String>() { // @Override // public String getKey(Tuple2<String, Integer> value) throws Exception { // return value.f0; // } // }); KeyedStream<Tuple2<String, Integer>, String> grouped = wordAndOne.keyBy(t -> t.f0); // 3.4:对分组后的key进行聚合操作 SingleOutputStreamOperator<Tuple2<String, Integer>> sumed = grouped.sum(1); //todo 4)构建sink,输出结果 sumed.print(); //todo 5)启动运行 env.execute(); } }
1上传作业jar包
2指定递交参数
3查看任务运行概述
4查看任务运行结果