Flink实现WordCount(实操详细步骤)
时间:2022-07-25
本文章向大家介绍Flink实现WordCount(实操详细步骤),主要内容包括其使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。
本文使用 Flink 的两种方式实现 WordCount
- 基于流计算
- 基于批计算
文章目录
- 1. Idea 新建 Maven 项目并配置以下依赖
- 2. 实现代码及详细注释
- 2.1 Flink 基于流计算实现 WordCount
- 2.2 Flink 基于批计算实现 WordCount
- 2.3 附件:完整代码
- 2.1 Flink 基于流计算实现 WordCount
- 2.2 Flink 基于批计算实现 WordCount
- 2.3 附件:完整代码
先说一下我的环境:
Flink 1.9
开发工具:Idea
Maven版本:3.3.9
Linux:CentOS 7
演示语言:Scala 2.11
1. Idea 新建 Maven 项目并配置以下依赖
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>1.9.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>1.9.1</version>
</dependency>
</dependencies>
2. 实现代码及详细注释
2.1 Flink 基于流计算实现 WordCount
案例需求:采用 Netcat 数据源发送数据,使用Flink统计每个单词的数量
Idea执行代码 –> 打开 Linux 使用 nc(netcat)命令发送数据测试
nc -lk 8888
2.2 Flink 基于批计算实现 WordCount
需求:读取本地数据文件,统计文件中每个单词出现的次数
wc.txt
文件的内容
hadoop hbase hello
hello hadoop apache apache
flink hello
执行代码结果
2.3 附件:完整代码
package com.bigdataBC.flink
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
/**
* 基于流计算的WordCount案例
*/
object WordCountBySrteaming {
def main(args: Array[String]): Unit = {
// 初始化Flink的Streaming(流计算)上下文执行环境
val streamEvn: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//导入隐式转换,建议写在这里,可以防止IDEA代码提示出错的问题
import org.apache.flink.streaming.api.scala._
// 设置默认的分区(分区优先级:先找单独设置的分区,若没有就用默认的)
streamEvn.setParallelism(1)
// 读取流数据
val ds: DataStream[String] = streamEvn.socketTextStream("node1",8888)
// 转换计算
val result: DataStream[(String, Int)] = ds.flatMap(_.split(" "))
.map((_, 1))
.setParallelism(2) //设置单独的分区
.keyBy(0) // 分组:必须制定根据哪个字段分组,参数代表当前要分组的字段的下标(另外还有fieldsname)
.sum(1) // 1代表下标,下标为1的进行累加
//打印结果到控制台
result.print()
.setParallelism(4) //设置单独的分区
//启动流式处理,如果没有该行代码上面的程序不会运行
streamEvn.execute("wordcount")
}
}
package com.bigdataBC.flink
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
/**
* 基于批计算的WordCount案例
*/
object WordCountByBatch {
def main(args: Array[String]): Unit = {
// 初始化Flink批计算环境、
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
// 导入隐式转换
import org.apache.flink.api.scala._
// 设置默认的分区
// env.setParallelism(1)
// 读取数据
val ds: DataSet[String] = env.readTextFile("D:\workspace\Idea-workspace\Flinkdemo\src\main\resources\wc.txt")
// 转换计算
val result: AggregateDataSet[(String, Int)] = ds.flatMap(_.split(" "))
.map((_, 1))
.groupBy(0)
.sum(1)
// 打印(这里的print不能设置分区)
result.print()
}
}
- JS游戏开发 可移动地图的实现
- Java面试系列-多线程
- pymongo.errors:Sort operation used more than the maximum 33554432 bytes of RAM. Add an index,
- PG学习初体验--源码安装和简单命令(r8笔记第97天)
- Pymongo: TypeError: if no direction is specified, key_or_list must be an instance of list
- Java面试系列19-Struts2
- JS游戏开发,让你的静态人物动起来(来自网路)
- Golang语言websocket源码
- 社团划分——Fast Unfolding算法
- 【PSU】AIX 11g RAC自动打GI PSU5
- 详解自动识别验证码,LSTM大显身手
- 智能算法——PageRank
- 【Go 语言社区】Golang 可变参数的使用
- 【Go 语言社区】Golang源码解读之map
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- Java中类型判断的几种方式
- 如何实现类似@Component的Spring动态注入功能
- 一次代码优化实践,用了模板方法+策略+工厂方法模式
- 给 JDK 报了一个 P4 的 Bug,结果居然……
- Git | Git Server 搭建
- leetcode树之二叉树的深度
- Nic*app的native层算法破解
- ActiveMq的顺序性消费问题
- 佳能遭严重勒索软件攻击,10TB数据被窃取,大量服务宕机
- 进程线程剖析(二)-进程组成、状态与特点
- 【Flink】基于 Flink 实时计算商品订单流失量
- 这就是你日日夜夜想要的docker!!!---------TLS加密远程连接Docker
- 【Flink】基于 Flink 的流式数据实时去重
- 从零开始安装穿透式检索
- 使用kind搭建kubernetes