个人网站备案名字大全,怎么做网站百度贴吧,用织梦系统做的2个网站要把它都上传到服务器上吗,计算机怎么建设网站Flink
Apache Flink — 数据流上的有状态计算。
Apache Flink 是一个框架和分布式处理引擎#xff0c;用于对无界和有界数据流进行有状态计算处理。
任何类型的数据都以事件流的形式生成。信用卡交易、传感器测量、机器日志或网站或移动应用程序 2上的用户交互#xff0c;…Flink
Apache Flink — 数据流上的有状态计算。
Apache Flink 是一个框架和分布式处理引擎用于对无界和有界数据流进行有状态计算处理。
任何类型的数据都以事件流的形式生成。信用卡交易、传感器测量、机器日志或网站或移动应用程序 2上的用户交互所有这些数据都以流的形式生成。
数据可以作为无界或有界流进行处理。
无界数据流有定义流的开始但是没有定义结束。会一直提供数据没有结束。所以要一直连续的处理无界流所以一旦有数据到来就要立即处理不能等数据都到再处理因为输入是无限的。处理无界数据通常需要按特定顺序如数据引入的顺序以便能够推断结果的完整性。
有界数据流有具体的开始和结束。有界流的处理也称为批处理。有界数据可以等待所有数据到达之后再进行计算处理。有界数据不需要按顺序引入因为可以对有界的数据集进行排序。 # 添加完这些依赖就可以使用Java代码使用Flink的流处理功能。# 这个依赖项包含了Flink的流处理API和相关的类库。主要干活的
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-java/artifactIdversion${flink.version}/version
/dependency# Flink客户端库用这个可以连接到Flink集群并提交或管理Flink作业。
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-clients/artifactIdversion${flink.version}/version
/dependencyFlink部署 flink部署模式
在一些应用场景中对于集群资源分配和占用的方式可能会有特定的需求。Flink为各种场景提供了不同的部署模式主要有以下三种会话模式Session Mode、单作业模式Per-Job Mode、应用模式Application Mode。
它们的区别主要在于集群的生命周期以及资源的分配方式以及应用的main方法到底在哪里执行——客户端Client还是JobManager。
会话模式Session Mode *0集群规划*
表3-1 集群角色分配
节点服务器*hadoop**102**hadoop**103**hadoop**104*角色充当JobManager和TaskManagerTaskManagerTaskManager
具体安装部署步骤如下
*1下载并解压安装包*
1https://flink.apache.org/downloads/ 下载安装包flink-1.17.0-bin-scala_2.12.tgz将该jar包上传到hadoop102节点服务器的/opt/software路径上。
2在/opt/software路径上解压flink-1.17.0-bin-scala_2.12.tgz到/opt/module路径上。
[atguiguhadoop102 software]$ tar -zxvf flink-1.17.0-bin-scala_2.12.tgz -C /opt/module/*2修改集群配置*
1进入conf路径修改flink-conf.yaml文件指定hadoop102节点服务器为JobManager
[atguiguhadoop102 conf]$ vim flink-conf.yaml修改如下内容
# JobManager节点地址.jobmanager.rpc.address: hadoop102jobmanager.bind-host: 0.0.0.0rest.address: hadoop102rest.bind-address: 0.0.0.0\# TaskManager节点地址.需要配置为当前机器名taskmanager.bind-host: 0.0.0.0taskmanager.host: hadoop1022修改workers文件指定hadoop102、hadoop103和hadoop104为TaskManager
[atguiguhadoop102 conf]$ vim workers修改如下内容
hadoop102hadoop103hadoop1043修改masters文件
[atguiguhadoop102 conf]$ vim masters修改如下内容
hadoop102:80814另外在flink-conf.yaml文件中还可以对集群中的JobManager和TaskManager组件进行优化配置主要配置项如下
l jobmanager.memory.process.size对JobManager进程可使用到的全部内存进行配置包括JVM元空间和其他开销默认为1600M可以根据集群规模进行适当调整。
l taskmanager.memory.process.size对TaskManager进程可使用到的全部内存进行配置包括JVM元空间和其他开销默认为1728M可以根据集群规模进行适当调整。
l taskmanager.numberOfTaskSlots对每个TaskManager能够分配的Slot数量进行配置默认为1可根据TaskManager所在的机器能够提供给Flink的CPU数量决定。所谓Slot就是TaskManager中具体运行一个任务所分配的计算资源。
l parallelism.defaultFlink任务执行的并行度默认为1。优先级低于代码中进行的并行度配置和任务提交时使用参数指定的并行度数量。
关于Slot和并行度的概念我们会在下一章做详细讲解。
*3分发安装目录*
1配置修改完毕后将Flink安装目录发给另外两个节点服务器。
[atguiguhadoop102 module]$ xsync flink-1.17.0/2修改hadoop103的 taskmanager.host
[atguiguhadoop103 conf]$ vim flink-conf.yaml修改如下内容
# TaskManager节点地址.需要配置为当前机器名taskmanager.host: hadoop1033修改hadoop104的 taskmanager.host
[atguiguhadoop104 conf]$ vim flink-conf.yaml修改如下内容
# TaskManager节点地址.需要配置为当前机器名taskmanager.host: hadoop104*4启动集群*
1在hadoop102节点服务器上执行start-cluster.sh启动Flink集群
[atguiguhadoop102 flink-1.17.0]$ bin/start-cluster.sh2查看进程情况
[atguiguhadoop102 flink-1.17.0]$ jpsall hadoop102
4453 StandaloneSessionClusterEntrypoint
4458 TaskManagerRunner
4533 Jps hadoop103
2872 TaskManagerRunner
2941 Jps hadoop104
2948 Jps
2876 TaskManagerRunner
*5访问Web UI*
启动成功后同样可以访问http://hadoop102:8081对flink集群和任务进行监控管理。
这里可以明显看到当前集群的TaskManager数量为3由于默认每个TaskManager的Slot数量为1所以总Slot数和可用Slot数都为3。
向集群提交作业
在上一章中我们已经编写读取socket发送的单词并统计单词的个数程序案例。本节我们将以该程序为例演示如何将任务提交到集群中进行执行。具体步骤如下。
*1**环境准备*
在hadoop102中执行以下命令启动netcat。
[atguiguhadoop102 flink-1.17.0]$ nc -lk 7777*2****程序**打包*
1在我们编写的Flink入门程序的pom.xml文件中添加打包插件的配置具体如下
buildpluginsplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-shade-plugin/artifactIdversion3.2.4/versionexecutionsexecutionphasepackage/phasegoalsgoalshade/goal/goalsconfigurationartifactSetexcludesexcludecom.google.code.findbugs:jsr305/excludeexcludeorg.slf4j:*/excludeexcludelog4j:*/exclude/excludes/artifactSetfiltersfilter!-- Do not copy the signatures in the META-INF folder.Otherwise, this might cause SecurityExceptions when using the JAR. --artifact*:*/artifactexcludesexcludeMETA-INF/*.SF/excludeexcludeMETA-INF/*.DSA/excludeexcludeMETA-INF/*.RSA/exclude/excludes/filter/filterstransformers combine.childrenappendtransformerimplementationorg.apache.maven.plugins.shade.resource.ServicesResourceTransformer/transformer/transformers/configuration/execution/executions/plugin/plugins/build2插件配置完毕后可以使用IDEA的Maven工具执行package命令出现如下提示即表示打包成功。
-------------------------------------------------------------------
[INFO] BUILD SUCCESS
-------------------------------------------------------------------
打包完成后在target目录下即可找到所需JAR包JAR包会有两个FlinkTutorial-1.0-SNAPSHOT.jar和FlinkTutorial-1.0-SNAPSHOT-jar-with-dependencies.jar因为集群中已经具备任务运行所需的所有依赖所以建议使用FlinkTutorial-1.0-SNAPSHOT.jar。比较大的带有依赖。
*3**在Web* *UI上提交作业*
1任务打包完成后我们打开Flink的WEB UI页面在右侧导航栏点击“Submit New Job”然后点击按钮“ Add New”选择要上传运行的JAR包如下图所示。
JAR包上传完成如下图所示
2点击该JAR包出现任务配置页面进行相应配置。
主要配置程序入口主类的全类名任务运行的并行度任务运行所需的配置参数和保存点路径等如下图所示配置完成后即可点击按钮“Submit”将任务提交到集群运行。
3任务提交成功之后可点击左侧导航栏的“Running Jobs”查看程序运行列表情况。
4测试
①在socket端口中输入hello
[atguiguhadoop102 flink-1.17.0]$ nc -lk 7777
hello
②先点击Task Manager然后点击右侧的192.168.10.104服务器节点
③点击Stdout就可以看到hello单词的统计
注意如果hadoop104节点没有统计单词数据可以去其他TaskManager节点查看。
4点击该任务可以查看任务运行的具体情况也可以通过点击“Cancel Job”结束任务运行。
*4**命令行提交作业*
除了通过WEB UI界面提交任务之外也可以直接通过命令行来提交任务。这里为方便起见我们可以先把jar包直接上传到目录flink-1.17.0下
1首先需要启动集群。
[atguiguhadoop102 flink-1.17.0]$ bin/start-cluster.sh
2在hadoop102中执行以下命令启动netcat。
[atguiguhadoop102 flink-1.17.0]$ nc -lk 7777
3将flink程序运行jar包上传到/opt/module/flink-1.17.0路径。
4进入到flink的安装路径下在命令行使用flink run命令提交作业。
[atguiguhadoop102 flink-1.17.0]$ bin/flink run -m hadoop102:8081 -c com.atguigu.wc.SocketStreamWordCount ./FlinkTutorial-1.0-SNAPSHOT.jar这里的参数 -m指定了提交到的JobManager-c指定了入口类。
5在浏览器中打开Web UIhttp://hadoop102:8081查看应用执行情况。
用netcat输入数据可以在TaskManager的标准输出Stdout看到对应的统计结果。
6在/opt/module/flink-1.17.0/log路径中可以查看TaskManager节点。
[atguiguhadoop102 log]$ cat flink-atguigu-standalonesession-0-hadoop102.out
(hello,1)
(hello,2)
(flink,1)
(hello,3)
(scala,1)
单作业模式Per-Job Mode 应用模式Application Mode 这里我们所讲到的部署模式相对是比较抽象的概念。实际应用时一般需要和资源管理平台结合起来选择特定的模式来分配资源、部署应用。接下来我们就针对不同的资源提供者的场景具体介绍Flink的部署方式。
DataStream API
DataStream API是Flink的核心层API使用API实现对数据流的计算和处理。
一个Flink程序其实就是对数据流DataStream的各种转换。具体来说代码基本上都由以下几部分构成
/*** 计算单词出现个数** flink处理无界数据流* 程序会一直运行一有数据来就处理** author shkstart* create 2023-09-10 16:44*/
public class SocketStreamWordCount {public static void main(String[] args) throws Exception {// 1.创建flink流式处理环境 StreamExecutionEnvironmentStreamExecutionEnvironment see StreamExecutionEnvironment.getExecutionEnvironment();// 2.接收要待处理的数据DataStreamSourceString dateStream see.socketTextStream(192.168.239.128, 7777);// 3.处理数据 数据处理后格式(word,2)单词和对应出现的次数/*** flatMap(FlatMapFunctionT, R flatMapper)* 为数据流的每一个元素调用flatMapper*/System.out.println(原始数据流 dateStream);// FlatMapFunction转换处理数据流元素FlatMapFunctionString, Tuple2String, Integer flatMapFunction new FlatMapFunctionImpl();SingleOutputStreamOperatorTuple2String, Integer transformedDataStream dateStream.flatMap(flatMapFunction);System.out.println(处理后的数据流 transformedDataStream);// 按照word分组 按string分组 将Integer累加SingleOutputStreamOperatorTuple2String, Integer sum transformedDataStream.keyBy(data - data.f0).sum(1);// 4.展示sum.print();// 5.执行 开始处理// 代码末尾需要调用 流式处理环境 的execute方法开始执行任务see.execute();}
}
public class FlatMapFunctionImpl implements FlatMapFunctionString , Tuple2String,Integer {/*** 转换数据流元素* param value 输入的元素* param out 输出的元素* throws Exception*/Overridepublic void flatMap(String value, CollectorTuple2String, Integer out) throws Exception {// 切分String[] words value.split( );// 收集for (String word : words) {out.collect(Tuple2.of(word,1));}}
}
1、执行环境
Flink程序可以在各种上下文环境中运行我们可以在本地JVM中执行程序也可以提交到远程集群上运行。
不同的环境代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时首先必须获取当前Flink的运行环境从而建立起与Flink框架之间的联系。
DataStream API执行模式包括流执行模式、批执行模式和自动模式。
流执行模式Streaming
这是DataStream API最经典的模式一般用于需要持续实时处理的无界数据流。默认情况下程序使用的就是Streaming执行模式。
// 流处理环境
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();批执行模式Batch
专门用于批处理处理有界数据的执行模式。
自动模式AutoMatic
在这种模式下将由程序根据输入数据源是否有界来自动选择执行模式。
批执行模式的使用。主要有两种方式
1通过命令行配置
bin/flink run -Dexecution.runtime-modeBATCH ...在提交作业时增加execution.runtime-mode参数指定值为BATCH。
2通过代码配置
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.BATCH);在代码中直接基于执行环境调用setRuntimeMode方法传入BATCH模式。
实际应用中一般不会在代码中配置而是使用命令行这样更加灵活。
2、数据源
从socket中读取数据
不论从集合还是文件我们读取的其实都是有界数据。在流处理的场景中数据往往是无界的。
我们之前用到的读取socket文本流就是流处理场景。但是这种方式由于吞吐量小、稳定性较差一般也是用于测试。
DataStreamString stream env.socketTextStream(localhost, 7777);从Kafka读取数据
Flink官方提供了连接工具flink-connector-kafka直接帮我们实现了一个消费者FlinkKafkaConsumer它就是用来读取Kafka数据的SourceFunction。
所以想要以Kafka作为数据源获取数据我们只需要引入Kafka连接器的依赖。Flink官方提供的是一个通用的Kafka连接器它会自动跟踪最新版本的Kafka客户端。目前最新版本只支持0.10.0版本以上的Kafka。这里我们需要导入的依赖如下。
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-kafka/artifactIdversion${flink.version}/version/dependency代码如下
public class SourceKafka {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();KafkaSourceString kafkaSource KafkaSource.Stringbuilder().setBootstrapServers(hadoop102:9092).setTopics(topic_1).setGroupId(atguigu).setStartingOffsets(OffsetsInitializer.latest()).setValueOnlyDeserializer(new SimpleStringSchema()) .build();DataStreamSourceString stream env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), kafka-source);stream.print(Kafka);env.execute();}
}3、转换算子
基本转换算子map/** filter**/** flat**Map****
