asp.net网站开发简介,物联网官网入口,网页设计师联盟qq,wordpress绝对连接换成相对链接一、运行环境介绍
Flink执行环境主要分为本地环境和集群环境#xff0c;本地环境主要为了方便用户编写和调试代码使用#xff0c;而集群环境则被用于正式环境中#xff0c;可以借助Hadoop Yarn、k8s或Mesos等不同的资源管理器部署自己的应用。
环境依赖#xff1a; 【1】…一、运行环境介绍
Flink执行环境主要分为本地环境和集群环境本地环境主要为了方便用户编写和调试代码使用而集群环境则被用于正式环境中可以借助Hadoop Yarn、k8s或Mesos等不同的资源管理器部署自己的应用。
环境依赖 【1】JDK环境Flink核心模块均使用 Java开发所以运行环境需要依赖JDKJDK版本需要保证在1.8以上。 【2】Maven编译环境Flink的源代码目前仅支持通过 Maven进行编译所以如果需要对源代码进行编译或通过IDE开发Flink Application则建议使用Maven作为项目工程编译方式。需要注意的是Flink程序需要Maven的版本在3.0.4及以上否则项目编译可能会出问题建议用户根据要求进行环境的搭建。 【3】IDEA需要安装scala插件以及scala环境等
二、Flink项目 Scala版 DataSet 有界流
需求同进文件文件中的单词出现的次数
【1】创建Maven项目pom.xml文件中配置如下依赖
dependenciesdependencygroupIdorg.apache.flink/groupIdartifactIdflink-scala_2.12/artifactIdversion1.10.0/version/dependency!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-scala_2.12/artifactIdversion1.10.0/version/dependency
/dependenciesbuildplugins!-- 该插件用于将Scala代码编译成class文件 --plugingroupIdnet.alchim31.maven/groupIdartifactIdscala-maven-plugin/artifactIdversion3.4.6/versionexecutionsexecutiongoals!--声明绑定到 maven 的compile阶段--goalcompile/goal/goals/execution/executions/pluginplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-assembly-plugin/artifactIdversion3.0.0/versionconfigurationdescriptorRefsdescriptorRefjar-with-dependencies/descriptorRef/descriptorRefs/configurationexecutionsexecutionidmake-assembly/idphasepackage/phasegoalsgoalsingle/goal/goals/execution/executions/plugin/plugins
/build【2】resource目录中添加需要进行统计的文件文件及内容
【3】WordCount.java文件内容如下需要注意隐私转换问题需要引入scala._ import org.apache.flink.api.scala._/**
* Description 批处理 word count
* Author zhengzhaoxiang
* Date 2020/7/12 18:55
* Param
* Return
*/
object WordCount {def main(args: Array[String]): Unit {//创建一个批处理的执行环境val env: ExecutionEnvironment ExecutionEnvironment.getExecutionEnvironment//从文件中读取数据var inputDateSet: DataSet[String] env.readTextFile(E:\\Project\\flink\\src\\main\\resources\\wordcount.txt)//基于Dataset 做转换首先按空格打散然后按照 word作为key做group byval resultDataSet: DataSet[(String,Int)] inputDateSet.flatMap(_.split( ))//分词得到所有 word构成的数据集.map((_,1))//_表示当前 word 转换成一个二元组word,count).groupBy(0)//以二元组中第一个元素作为key.sum(1) //1表示聚合二元组的第二个元素的值//打印输出resultDataSet.print()}
}【4】统计结果展示
三、Flink项目 Scala版 DataStream 无界流
【1】StreamWordCount.java文件内容如下
package com.zzx.flinkimport org.apache.flink.streaming.api.scala._object StreamWordCount {def main(args: Array[String]): Unit {// 创建一个流处理执行环境val env: StreamExecutionEnvironment StreamExecutionEnvironment.getExecutionEnvironment// 接受 socket 文本流val inputDataStream: DataStream[String] env.socketTextStream(hadoop1,6666);//定义转换操作 word countval resultDataStream: DataStream[(String,Int)] inputDataStream.flatMap(_.split( ))//以空格分词得到所有的 word.filter(_.nonEmpty).map((_,1))//转换成 word count 二元组.keyBy(0)//按照第一个元素分组.sum(1)//按照第二个元素求和resultDataStream.print()//上面的只是定义了处理流程同时定义一个名称。不会让任务结束env.execute(stream word count word)}
}【2】我这里在Hadoop1中通过nc -lk xxx打开一个socket通信
【3】查看IDEA输出统计内容如下输出word的顺序不是按照输入的顺序是因为它有并行度多线程是并行执行的。最前面的数字是并行子任务的编号类似线程号。最大的数字其实跟你cpu核数是息息相关的。这个并行度也可以通过env.setParallelism进行设置。我们也可以给每一个任务算子设置不同的并行度
【4】当我们需要将Java文件打包上传到Flink的时候这里的host和port可以从参数中进行获取代码修改如下
package com.zzx.flinkimport org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._object StreamWordCount {def main(args: Array[String]): Unit {// 创建一个流处理执行环境val env: StreamExecutionEnvironment StreamExecutionEnvironment.getExecutionEnvironment// 接受 socket 文本流 hostname:prot 从程序运行参数中读取val params: ParameterTool ParameterTool.fromArgs(args);val hostname: String params.get(host);val port: Int params.getInt(port);val inputDataStream: DataStream[String] env.socketTextStream(hostname,port);//定义转换操作 word countval resultDataStream: DataStream[(String,Int)] inputDataStream.flatMap(_.split( ))//以空格分词得到所有的 word.filter(_.nonEmpty).map((_,1))//转换成 word count 二元组.keyBy(0)//按照第一个元素分组.sum(1)//按照第二个元素求和resultDataStream.print()//上面的只是定义了处理流程同时定义一个名称。不会让任务结束env.execute(stream word count word)}
}
