207.Flink（二）：架构及核心概念，flink从各种数据源读取数据，各种算子转化数据，将数据推送到各数据源

一、Flink架构及核心概念

1.系统架构

  • JobMaster是JobManager中最核心的组件，负责处理单独的作业（Job）。
  • 一个job对应一个jobManager

 2.并行度

（1）并行度（Parallelism）概念

一个特定算子的子任务（subtask）的个数被称之为其并行度（parallelism）。这样，包含并行子任务的数据流，就是并行数据流，它需要多个分区（stream partition）来分配并行任务。

流程序的并行度 = 其所有算子中最大的并行度。一个程序中，不同的算子可能具有不同的并行度。

（2）设置并行度

对某个具体算子设置并行度：

stream.map(word -> Tuple2.of(word, 1L)).setParallelism(2);

全局设置并行度：

env.setParallelism(2);

提交任务时指定：

  • 通过页面上传jar的时候可以指定
  • 可以在命令行启动的时候通过 -p 3指定

flink-conf.yaml中配置：

parallelism.default: 2

优先级：

代码中具体算子 > 代码中全局 > 提交任务指定 > 配置文件中指定

3.算子链

（1）算子间的数据传输

*1）一对一（One-to-one，forwarding）

这种模式下，数据流维护着分区以及元素的顺序。它们之间不需要重新分区，也不需要调整数据的顺序。map、filter、flatMap等算子都是这种one-to-one的对应关系。这种关系类似于Spark中的窄依赖。

*2）重分区（Redistributing）

在这种模式下，数据流的分区会发生改变。每一个算子的子任务，会根据数据传输的策略，把数据发送到不同的下游目标任务。这些传输方式都会引起重分区的过程，这一过程类似于Spark中的shuffle。

（2）合并算子链

在Flink中，并行度相同的一对一（one to one）算子操作，可以直接链接在一起形成一个“大”的任务（task），这样原来的算子就成为了真正任务里的一部分 

// 禁用算子链,该算子不会和前面和后面串在一起
.map(word -> Tuple2.of(word, 1L)).disableChaining();// 全局禁用算子链
env.disableChaining();// 从当前算子开始新链
.map(word -> Tuple2.of(word, 1L)).startNewChain()

  • 当一对一的时候，每个运算量都很大，这个时候不适合串在一起。
  • 当需要定位具体问题的时候，不串在一起更容易排查问题

4.任务槽

（1）任务槽（Task Slots）概念

Flink中每一个TaskManager都是一个JVM进程，它可以启动多个独立的线程，来并行执行多个子任务（subtask）。

TaskManager的计算资源是有限的，为了控制并发量，TaskManager对每个任务运行所占用的内存资源做出明确的划分，这就是所谓的任务槽（task slots）。

每个任务槽的大小是均等的，且任务槽之间的资源不可以互相借用。

如图，每个TaskManager有三个任务槽，每个槽运行自己的任务。槽的大小均等。

（2）任务槽数量的设置

在Flink的/opt/module/flink-1.17.0/conf/flink-conf.yaml配置文件中，可以设置TaskManager的slot数量，默认是1个slot。

taskmanager.numberOfTaskSlots: 8

slot目前仅仅用来隔离内存，不会涉及CPU的隔离。在具体应用时，建议将slot数量配置为机器的CPU核心数。

（3）任务对任务槽的共享

在同一个作业中，不同任务节点的并行子任务可以放在同一个slot上执行

 可以共享：

  • 同一个job中，不同算子的子任务才可以共享同一个slot。这些子任务是同时运行
  • 前提是：属于同一个slot共享组，默认都是“default”

手动指定共享组：

.map(word -> Tuple2.of(word, 1L)).slotSharingGroup("1");

共享的好处：允许我们保存完整的作业管道。这样一来，即使某个TaskManager出现故障宕机，其他节点也可以完全不受影响，作业的任务可以继续执行

（4）任务槽和并行度的关系

  • 任务槽是静态的概念，是指TaskManager具有的并发执行能力，可以通过参数taskmanager.numberOfTaskSlots进行配置
  • 并行度是动态概念，也就是TaskManager运行程序时实际使用的并发能力，可以通过参数parallelism.default进行配置

如果是yarn模式，申请的TaskManager的数量 = job并行度 / 每个TM的slot数量，向上取整

即：假设10个并行度的job，每个TM的slot是3个，那么需要10/3,向上取整，即需要最少4个TaskManager

二、作业提交流程

1.Standalone会话模式作业提交流程

逻辑流图（StreamGraph）→ 作业图（JobGraph）→ 执行图（ExecutionGraph）→ 物理图（Physical Graph）。

  • 逻辑流图：列出并行度，算子，各算子之间关系（一对一还是需要重分区）
  • 作业图：将一对一的算子做算子链的优化，作业中间会有中间结果集
  • 执行图：将并行度展开，并标注每个并行处理的算子
  • 物理图：基本同执行图，是执行图的落地

2.Yarn应用模式作业提交流程

三、 DataStream API

DataStream API是Flink的核心层API。一个Flink程序，其实就是对DataStream的各种转换。

1.执行环境（Execution Environment）

（1）创建执行环境

*1）StreamExecutionEnvironment.getExecutionEnvironment();

它会根据当前运行的上下文直接得到正确的结果：如果程序是独立运行的，就返回一个本地执行环境；如果是创建了jar包，然后从命令行调用它并提交到集群执行，那么就返回集群的执行环境

*2）StreamExecutionEnvironment.createLocalEnvironment();

这个方法返回一个本地执行环境。可以在调用时传入一个参数，指定默认的并行度；如果不传入，则默认并行度就是本地的CPU核心数

*3）StreamExecutionEnvironment
          .createRemoteEnvironment(
            "host",                   // JobManager主机名
            1234,                     // JobManager进程端口号
               "path/to/jarFile.jar"  // 提交给JobManager的JAR包
        );

这个方法返回集群执行环境。需要在调用时指定JobManager的主机名和端口号，并指定要在集群中运行的Jar包。

 （2）执行模式（Execution Mode）

流批一体：代码api是同一套，可以指定为 批，也可以指定为 流。

通话代码配置：

env.setRuntimeMode(RuntimeExecutionMode.BATCH);

通过命令行配置：

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

（3）触发程序执行

当main()方法被调用时，并没有真正处理数据。只有等到数据到来，才会触发真正的计算，这也被称为“延迟执行”或“懒执行”。

所以我们需要显式地调用执行环境的execute()方法，来触发程序执行。execute()方法将一直等待作业完成，然后返回一个执行结果（JobExecutionResult）。

如果在一段代码里面执行多个任务，可以使用env.executeAsync();

package com.atguigu.env;import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;/*** TODO** @author cjp* @version 1.0*/
public class EnvDemo {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.set(RestOptions.BIND_PORT, "8082");StreamExecutionEnvironment env = StreamExecutionEnvironment
//                .getExecutionEnvironment();  // 自动识别是 远程集群 ，还是idea本地环境.getExecutionEnvironment(conf); // conf对象可以去修改一些参数//                .createLocalEnvironment()
//        .createRemoteEnvironment("hadoop102", 8081,"/xxx")// 流批一体：代码api是同一套，可以指定为 批，也可以指定为 流// 默认 STREAMING// 一般不在代码写死，提交时 参数指定：-Dexecution.runtime-mode=BATCHenv.setRuntimeMode(RuntimeExecutionMode.BATCH);env
//                .socketTextStream("hadoop102", 7777).readTextFile("input/word.txt").flatMap((String value, Collector<Tuple2<String, Integer>> out) -> {String[] words = value.split(" ");for (String word : words) {out.collect(Tuple2.of(word, 1));}}).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(value -> value.f0).sum(1).print();env.execute();/** TODO 关于execute总结(了解)*     1、默认 env.execute()触发一个flink job：*          一个main方法可以调用多个execute，但是没意义，指定到第一个就会阻塞住*     2、env.executeAsync()，异步触发，不阻塞*         => 一个main方法里 executeAsync()个数 = 生成的flink job数*     3、思考：*         yarn-application 集群，提交一次，集群里会有几个flink job？*         =》 取决于 调用了n个 executeAsync()*         =》 对应 application集群里，会有n个job*         =》 对应 Jobmanager当中，会有 n个 JobMaster*/
//        env.executeAsync();// ……
//        env.executeAsync();}
}

2.源算子（Source）

从Flink1.12开始，主要使用流批统一的新Source架构：

DataStreamSource<String> stream = env.fromSource(…)

（1）创建pojo对象

需要空参构造器，所有属性的类型都是可以序列化的

package com.atguigu.bean;import java.util.Objects;/*** TODO** @author cjp* @version 1.0*/
public class WaterSensor {public String id;//水位传感器类型public Long ts;//传感器记录时间戳public Integer vc;//水位记录// 一定要提供一个 空参 的构造器public WaterSensor() {}public WaterSensor(String id, Long ts, Integer vc) {this.id = id;this.ts = ts;this.vc = vc;}public String getId() {return id;}public void setId(String id) {this.id = id;}public Long getTs() {return ts;}public void setTs(Long ts) {this.ts = ts;}public Integer getVc() {return vc;}public void setVc(Integer vc) {this.vc = vc;}@Overridepublic String toString() {return "WaterSensor{" +"id='" + id + '\'' +", ts=" + ts +", vc=" + vc +'}';}@Overridepublic boolean equals(Object o) {if (this == o) {return true;}if (o == null || getClass() != o.getClass()) {return false;}WaterSensor that = (WaterSensor) o;return Objects.equals(id, that.id) &&Objects.equals(ts, that.ts) &&Objects.equals(vc, that.vc);}@Overridepublic int hashCode() {return Objects.hash(id, ts, vc);}
}

（2）从集合中读取数据

package com.atguigu.source;import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** TODO** @author cjp* @version 1.0*/
public class CollectionDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// TODO 从集合读取数据DataStreamSource<Integer> source = env.fromElements(1,2,33); // 从元素读
//                .fromCollection(Arrays.asList(1, 22, 3));  // 从集合读source.print();env.execute();}
}

（3）从文件读取数据

先添加配置：

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-files</artifactId><version>1.17.0</version></dependency>
package com.atguigu.source;import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.environment.StreamExecutionEnvironment;/*** TODO** @author cjp* @version 1.0*/
public class FileSourceDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// TODO 从文件读： 新Source架构FileSource<String> fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(),new Path("input/word.txt")).build();env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "filesource").print();env.execute();}
}
/**** 新的Source写法：*   env.fromSource(Source的实现类，Watermark，名字)**/

（4）从Socket读取数据

DataStream<String> stream = env.socketTextStream("localhost", 7777);

（5）从Kafka读取数据

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>1.17.0</version>
</dependency>
package com.atguigu.source;import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.time.Duration;/*** TODO** @author cjp* @version 1.0*/
public cl

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.xdnf.cn/news/139599.html

如若内容造成侵权/违法违规/事实不符，请联系一条长河网进行投诉反馈，一经查实，立即删除！

相关文章

【方案】浅析AI视频分析与视频监控技术的工厂车间智能化监管方案

【方案】浅析AI视频分析与视频监控技术的工厂车间智能化监管方案

一、方案背景 工厂生产车间一般是从原材料到成品的流水作业&#xff0c;有大量器械和物料。为保障车间财产安全并提高生产效率&#xff0c;需要进行全面的监管。在生产制造流水线的关键工序中&#xff0c;不仅有作业过程监管需求&#xff0c;同时&#xff0c;也存在生产发生异…
阅读更多...
第七章 查找

第七章 查找

一、树形查找-二叉排序树和红黑树 二叉排序树 // 二叉排序树节点 typedef struct BSTNode{ElemType key;struct BSTNode *lchild, *rchild; } BSTNode, *BSTree;五叉查找树 // 5叉排序树的节点定义 struct Node{ElemType keys[4]; // 5叉查找树一个节点最多4个关键字struct…
阅读更多...
【二、安装centOS】

【二、安装centOS】

下载 地址&#xff1a;https://mirrors.aliyun.com/centos/ 地址 1、https://mirrors.aliyun.com/centos/7.9.2009/ 2、https://mirrors.aliyun.com/centos/7.9.2009/isos/ 3、https://mirrors.aliyun.com/centos/7.9.2009/isos/x86_64/ 选哪一个 可以选择第一个&#xff0…
阅读更多...
Openresty(二十二)ngx.balance和balance_by_lua终结篇

Openresty(二十二)ngx.balance和balance_by_lua终结篇

一 灰度发布铺垫 ① init_by_lua* init_by_lua init_by_lua_block 特点: 在openresty start、reload、restart时执行,属于master init 阶段机制&#xff1a; nginx master 主进程加载配置文件时&#xff0c;运行全局Lua VM级别上的参数指定的Lua代码场景&#xff1a; …
阅读更多...
java版Spring Cloud+Mybatis+Oauth2+分布式+微服务+实现工程管理系统

java版Spring Cloud+Mybatis+Oauth2+分布式+微服务+实现工程管理系统

鸿鹄工程项目管理系统 Spring CloudSpring BootMybatisVueElementUI前后端分离构建工程项目管理系统 1. 项目背景 一、随着公司的快速发展&#xff0c;企业人员和经营规模不断壮大。为了提高工程管理效率、减轻劳动强度、提高信息处理速度和准确性&#xff0c;公司对内部工程管…
阅读更多...
（二）随机变量的数字特征：探索概率分布的关键指标

（二）随机变量的数字特征：探索概率分布的关键指标

文章目录 &#x1f34b;1. 随机变量的数学期望&#x1f34b;1.1 离散型随机变量的数学期望&#x1f34b;1.2 连续型随机变量的数学期望 &#x1f34b;2. 随机变量函数的数学期望&#x1f34b;2.1 一维随机变量函数的数学期望&#x1f34b;2.2 二维随机变量函数的数学期望 &…
阅读更多...
基于SpringBoot的网上超市系统的设计与实现

基于SpringBoot的网上超市系统的设计与实现

目录 前言 一、技术栈 二、系统功能介绍 管理员功能实现 用户功能实现 三、核心代码 1、登录模块 2、文件上传模块 3、代码封装 前言 网络技术和计算机技术发展至今&#xff0c;已经拥有了深厚的理论基础&#xff0c;并在现实中进行了充分运用&#xff0c;尤其是基于计…
阅读更多...
linux升级glibc-2.28

linux升级glibc-2.28

1.准备工作 1.1升级gcc到gcc8 # 安装devtoolset-8-gcc yum install centos-release-scl yum install devtoolset-8 scl enable devtoolset-8 -- bash# 启用工具 source /opt/rh/devtoolset-8/enable # 安装GCC-8 yum install -y devtoolset-8-gcc devtoolset-8-gcc-c devtoolse…
阅读更多...
open62541开发：添加sqlite3 历史数据库

open62541开发：添加sqlite3 历史数据库

历史数据库在OPCUA 应用中十分重要&#xff0c;例如OPCUA 网关和OPCUA 汇聚服务器中都需要历史数据库功能。但是open62541 协议栈中仅包含了基于内存的历史数据库&#xff0c;在实际应用中是不够的。本博文讨论open62541 中添加sqlite3 为基础的历史数据库若干问题。 借鉴 Gi…
阅读更多...
某手新版本sig3参数算法还原

某手新版本sig3参数算法还原

Frida Native层主动调用 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81…
阅读更多...
Lua学习笔记：探究package

Lua学习笔记：探究package

前言 本篇在讲什么 理解Lua的package 本篇需要什么 对Lua语法有简单认知 对C语法有简单认知 依赖Visual Studio工具 本篇的特色 具有全流程的图文教学 重实践&#xff0c;轻理论&#xff0c;快速上手 提供全流程的源码内容 ★提高阅读体验★ &#x1f449; ♠ 一级…
阅读更多...
ruoyi框架修改左侧菜单样式

ruoyi框架修改左侧菜单样式

菜单效果 ruoyi前端框架左侧的菜单很丑&#xff0c;我们需要修改一下样式&#xff0c;下面直接看效果。 修改代码 1、sidebar.scss .el-menu-item, .el-submenu__title {overflow: hidden !important;text-overflow: ellipsis !important;white-space: nowrap !important;//…
阅读更多...
微软宣布推广数字助理 Copilot；GPT 应用开发和思考

微软宣布推广数字助理 Copilot；GPT 应用开发和思考

&#x1f989; AI新闻 &#x1f680; 微软宣布推广基于生成式人工智能的数字助理 Copilot 摘要&#xff1a;微软宣布将基于生成式人工智能的数字助理 Copilot 推广到更多软件产品中。新的 AI 助理 Microsoft Copilot 将在 Windows 中无缝可用&#xff0c;包括 Windows 11 桌面…
阅读更多...
【C++面向对象侯捷】12.虚函数与多态 | 13.委托相关设计【设计模式 经典做法，类与类之间关联起来，太妙了，不断的想，不断的写代码】

【C++面向对象侯捷】12.虚函数与多态 | 13.委托相关设计【设计模式 经典做法，类与类之间关联起来，太妙了，不断的想，不断的写代码】

文章目录 12.虚函数与多态举例&#xff1a;委托 继承【观察者模式】13.委托相关设计Composite 组合模式Prototype 原型模式 12.虚函数与多态 纯虚函数 一定要 子类重新定义的 继承和复合 关系下的构造和析构 举例&#xff1a;委托 继承【观察者模式】 13.委托相关设计 问题…
阅读更多...
BOM与DOM--记录

BOM与DOM--记录

BOM基础&#xff08;BOM简介、常见事件、定时器、this指向&#xff09; BOM和DOM的区别和联系 JavaScript的DOM与BOM的区别与用法详解 DOM和BOM是什么&#xff1f;有什么作用&#xff1f; 图解BOM与DOM的区别与联系 BOM和DOM详解 JavaScript 中的 BOM&#xff08;浏览器对…
阅读更多...
蓝桥杯 题库 简单 每日十题 day8

蓝桥杯 题库 简单 每日十题 day8

01 扫雷 题目描述 在一个n行列的方格图上有一些位置有地雷&#xff0c;另外一些位置为空。 请为每个空位置标一个整数&#xff0c;表示周围八个相邻的方格中有多少个地雷。 输入描述 输入的第一行包含两个整数n&#xff0c;m。 第2行到第n1行每行包含m个整数&#xff0c;相邻整…
阅读更多...
数据结构——单链表

数据结构——单链表

目录 一.前言 二.链表表示和实现&#xff08;单链表&#xff09; 1.1 顺序表的优缺点 1.2 链表的概念及结构 1.3 打印函数 1.4 空间函数 1.5 尾插函数&#xff08;最最最麻烦的&#xff09; 1.5.1 尾插最关键部分&#xff01; 1.6 头插函数 1.7 尾删函数…
阅读更多...
深拷贝和浅拷贝

深拷贝和浅拷贝

两者区别 深拷贝和浅拷贝是用来描述对象或者对象数组这种引用数据类型的复制场景的。 浅拷贝 浅拷贝&#xff0c;就是只复制某个对象的指针&#xff0c;而不复制对象本身。这种复制方式意味着两个引用指针指向被复制对象的同一块内存地址&#xff0c;如下图&#xff1a; 深拷贝…
阅读更多...
conda的安装和使用

conda的安装和使用

参考资料&#xff1a; https://www.bilibili.com/read/cv8956636/?spm_id_from333.999.0.0 https://www.bilibili.com/video/BV1Mv411x775/?spm_id_from333.999.0.0&vd_source98d31d5c9db8c0021988f2c2c25a9620 目录 conda是啥以及作用conda的安装conda的启动conda的配置…
阅读更多...
git和github的入门操作

git和github的入门操作

之前因为工作中用的都是SVN版本控制工具&#xff0c;没接触过git和github&#xff0c;现在开始深入自学Django框架技术后&#xff0c;看到官网推荐使用git&#xff0c;然后这两天网上查阅了很多文章教程&#xff0c;学到入门操作需要学习的点&#xff0c;太多的知识点要后面慢慢…
阅读更多...
最新文章

Copyright @ 2022~2023