Flink实现WordCount（实操详细步骤）（转载）_flink流式处理使用流处理完成wordcount,可以采用netcat生成数据

本文使用 Flink 的两种方式实现 WordCount

基于流计算
基于批计算
文章目录

1. Idea 新建 Maven 项目并配置以下依赖
2. 实现代码及详细注释
   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 项目并配置以下依赖
   
   
   org.apache.flink
   flink-scala_2.11
   1.9.1
   
   
   org.apache.flink
   flink-streaming-scala_2.11
   1.9.1
   
   
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
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
1
2
3
执行代码结果
在这里插入图片描述

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”)

}
}
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

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()

}
}

原文链接：https://blog.csdn.net/weixin_44285445/article/details/107920077?utm_medium=distribute.pc_feed.none-task-blog-personrec_tag-9.nonecase&depth_1-utm_source=distribute.pc_feed.none-task-blog-personrec_tag-9.nonecase&request_id=5f345ea9df0ade6351323983