从零开发flink-01: 编写wordcount

1. 关于

官方文档
https://ci.apache.org/projects/flink/flink-docs-release-1.11/try-flink/local_installation.html
学习视频
https://www.bilibili.com/video/BV197411M7cQ?p=8

使用目的

智能推荐

ETL 数据清理

恶意操作判断

实时推荐

流数据分析

实时报表确保数据不丢失

复杂事件处理 (实时判断用户一系列操作满足什么条件会发生什么)

优势

高吞吐 (每秒处理量) 低延时 (内存保存上下文状态没所以低延时) 高性能

支持事件时间(避免因为数据延时等操作有误)

有状态管理

窗口 (支持时间条数会话窗口) (比如一分钟内的数据, 100条数据, 会话中的数据)

分布式快照(checkpoint实现容错) 自动开启

基于JVM实现独立内存管理 (减少JVM的GC影响 / 二进制存储降低存储大小)

savepoint 保存点手动开启, (比如停机维护时使用)

2 说明

版本

flink 1.11.1

scala 2.11

注意

本文参考了上面的视频学习链接,但是视频中的版本是1.9.1 和现在最新的1.11.1有一些不同的地方,所以把踩坑的记录下来.

3 测试项目1 (流计算)

参考官方文档 https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/project-configuration.html

下面分别写上流处理和批处理的wordcount代码

3.1 创建项目

使用idea创建maven项目
添加pom.xml中的依赖,并刷新maven下载

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.11</artifactId>
            <version>1.11.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>1.11.0</version>
        </dependency>
    </dependencies>

3.2 编写流计算的wordcount

├── pom.xml
├── src
│   ├── main
│   │   ├── resources
│   │   └── scala
│   │       └── com
│   │           └── cowkeys
│   │               └── FlinkStreamWordCount.scala
│   └── test
│       └── scala

FlinkStreamWordCount.scala

package com.cowkeys

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object FlinkStreamWordCount {
  def main(args: Array[String]): Unit = {
    // 1. 初始流计算的环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 2. 导入隐式转换
    import org.apache.flink.streaming.api.scala._
    // 3. 读取数据
    val stream: DataStream[String] = streamEnv.socketTextStream("localhost", 8888)
    // 4. 转换和处理数据
    val result: DataStream[(String, Int)] = stream.flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(0)
      .sum(1)
    // 5. 打印结果
    result.print("res")
    // 6. 启动流计算程序
    streamEnv.execute("wordcount")
  }
}

控制台使用nc 启动一个socket服务监听8888端口

> nc -lk 8888

在项目中右键 Run 运行 (这里出现了一些报警信息,但是不用管,程序已经运行了)

/usr/lib/jvm/java-1.8.0-openjdk-amd64/bin/java ...
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

在nc终端输入测试数据两行

> nc -lk 8888
flink apache count
flink apache

项目中会有下面的信息

/usr/lib/jvm/java-1.8.0-openjdk-amd64/bin/java ...
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
res:10> (flink,1)
res:11> (count,1)
res:10> (apache,1)
res:10> (flink,2)
res:10> (apache,2

3.3 编写批计算的wordcount

新建scala object 在同级目录下 FlinkBatchWordCount.scala

├── flinkTest.iml
├── pom.xml
├── src
│   ├── main
│   │   ├── resources
│   │   │   └── word.txt
│   │   └── scala
│   │       └── com
│   │           └── cowkeys
│   │               ├── FlinkBatchWordCount.scala
│   │               └── FlinkStreamWordCount.scala
│   └── test
│       └── scala

在resource目录新建word.txt,随便写入一些重复的单词

flink apache batch
operate word analysis
hello flink
word count core

FlinkBatchWordCount.scala 代码

package com.cowkeys
import java.net.URL
import org.apache.flink.api.scala.ExecutionEnvironment
/*
flink 批计算
*/
object FlinkBatchWordCount {
  def main(args: Array[String]): Unit = {
    // 初始化flink批处理环境
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    // 导入隐式转换
    import org.apache.flink.api.scala._

    // 获取文件路径,读取文件
    val dataPath: URL = getClass.getResource("/word.txt")
    val data: DataSet[String] = env.readTextFile(dataPath.getPath)

    // 计算
    data.flatMap(_.split(" "))
      .map((_,1))
      .groupBy(0)
      .sum(1)
      .print()
  }
}

右键运行, 执行成功

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
(count,1)
(batch,1)
(apache,1)
(flink,2)
(core,1)
(analysis,1)
(word,2)
(operate,1)
(hello,1)