flink入门案例之WordCount

flink入门案例之WordCount,以下测试代码都是在本地执行的

添加依赖

添加maven依赖

<dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-scala_2.11</artifactId><version>1.7.2</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.11</artifactId><version>1.7.2</version></dependency></dependencies>

build信息

<build><plugins><!-- 该插件用于将Scala代码编译成class文件 --><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.4.6</version><executions><execution><!-- 声明绑定到maven的compile阶段 --><goals><goal>testCompile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>3.0.0</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins>
</build>

批处理

import org.apache.flink.api.scala.{AggregateDataSet, DataSet, ExecutionEnvironment, createTypeInformation}object WordCount {def main(args: Array[String]): Unit = {//获取执行环境//创建一个执行环境，该环境代表当前在其中执行程序的上下文。// 如果程序是独立调用的，则此方法返回本地执行环境。// 如果从命令行客户端内部调用该程序以将其提交给集群，则此方法将返回该集群的执行环境。val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment//创建通过逐行读取给定文件而产生的字符串数据集。val inputDS: DataSet[String] = env.readTextFile("D:/test/a.txt")val wordCounts: AggregateDataSet[(String, Int)] = inputDS.flatMap(_.split(" ")).map((_, 1)).groupBy(0).sum(1)//将DataSet中的元素打印到调用print（）方法的JVM的标准输出流System.out中。// 对于在群集中执行的程序，此方法需要将DataSet的内容收集回客户端，以在客户端打印。//为每个元素编写的字符串由AnyRef.toString方法定义。//与（）和（）方法类似，此方法立即触发程序执行。wordCounts.print()}
}

流处理

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}object StreamWordCount {def main(args: Array[String]): Unit = {//替换为你自己的IPval host = "192.168.xx.1x1"val port = 9000// 创建流处理环境val env = StreamExecutionEnvironment.getExecutionEnvironment// 接收socket文本流val textDstream: DataStream[String] = env.socketTextStream(host, port)// flatMap和Map需要引用的隐式转换import org.apache.flink.api.scala._val value: DataStream[(String, Int)] = textDstream.flatMap(_.split("\\s")).filter((_.nonEmpty)).map((_, 1)).keyBy(0).sum(1)value.print().setParallelism(1)env.execute("Socket WordCountTest")}
}

之后在linux上用如下命令进行测试

[root@mypc01 ~]# nc -lk 9000

总结

导入包的时候一定要留意导入的是java的包还是scala的包,很多函数是两个包都有的,导入错了让你怀疑人生

flink入门案例之WordCount相关推荐

09_Flink入门案例、word-count程序（java和scala版本）、添加依赖、Flink Streaming和Batch的区别、在集群上执行程序等
1.9.Flink入门案例-wordCount 1.9.1.开发工具 1.9.2.编写java版本word-count程序 1.9.2.1.添加Flink Maven依赖 1.9.2.2.编写word ...
2021年大数据Flink（八）：Flink入门案例
目录 Flink入门案例前置说明 API 编程模型准备工程 pom文件 log4j.properties Flink初体验需求编码步骤代码实现 Flink入门案例前置说明 API API ...
【Spark分布式内存计算框架——Structured Streaming】3. Structured Streaming —— 入门案例：WordCount
1.3 入门案例:WordCount 入门案例与SparkStreaming的入门案例基本一致:实时从TCP Socket读取数据(采用nc)实时进行词频统计WordCount,并将结果输出到控制台C ...
Flink教程（04）- Flink入门案例
文章目录 01 引言 02 开发前准备 2.1 API 2.2 编程模型 03 入门案例 3.1 项目搭建 3.2 代码实现 3.2.1 基于DataSet 3.2.2 基于DataStream 3. ...
SparkStreaming 入门案例之wordcount
案例概述以nc作为源发送数据案例演示创建nc源,用于发送数据. [root@mypc01 ~]# nc -lk mypc01 10086 创建maven工程,导入依赖 <dependenc ...
updateStateByKey算子入门案例之wordCount
概念有一个参数,是个函数,该函数有两个参数,第一个是序列类型,第二个是Option类型 def updateStateByKey[S : ClassTag](updateFunc: (Seq[V], ...
Structured Streaming 入门案例之WordCount
1.编写一个流式计算的应用, 不断的接收外部系统的消息 2.对消息中的单词进行词频统计 3.统计全局的结果步骤 Socket Server 等待 Structured Streaming 程序连接 ...
Hadoop入门案例WordCount
wordcount可以说是hadoop的入门案例,也是基础案例主要体现思想就是mapreduce核心思想原始文件为hadoop.txt,内容如下: hello,java hello,java,li ...
Flink——入门WordCount程序
Flink是什么? Apache Flink is a framework and distributed processing engine for stateful computations ov ...

flink入门案例之WordCount

添加依赖

批处理

流处理

总结

flink入门案例之WordCount相关推荐

最新文章

热门文章