Flink 简介与环境配置

实验介绍

在学习一门新的技术之前，我们首先要了解它的历史渊源，也就是说它为什么会出现，它能够解决什么业务痛点。所以本节我们的学习目的是了解 Flink 的背景，并运行第一个 Flink 程序，对它有一个初步的印象。

知识点

• 流处理概述
• Flink 简介
• Flink 批处理 WordCount
• Flink 流处理 WordCount

流处理简介

流处理并不是一个新概念，但是要做好并不是一件容易的事情。提到流处理，我们最先想到的可能是金融交易、信号检测以及地图导航等领域的应用。但是近年来随着信息技术的发展，除了前面提到的三个领域，其它方向对数据时效性的要求也越来越高。随着 Hadoop 生态的崛起，Storm、Spark Streaming、Samza、MillWheel 等一众流处理技术开始走入大众视野，但是我们最熟悉的应该还是 Storm 和 Spark Streaming。

我们知道，“高吞吐”、“低延迟”和”exactly-once“是衡量一个流处理框架的重要指标。 Storm 虽然提供了低延迟的流处理，但是在高吞吐方面的表现并不算佳，可以说基本满足不了日益暴涨的数据量，而且也没办法保证精准一次消费。Spark Streaming 中通过微批次的批处理来模拟流处理，只要将批处理的批次分的足够小，那么从宏观上来看就是流处理，这也是 Spark Streaming 的核心思想。通过微观批处理的方式，Spark Streaming 也实现了高吞吐和 exactly-once 语义，时效性也有了大幅提升，在很长一段时间里占据流处理榜首。但是受限于其实现方式，依然存在几秒的延迟，对于那些实时性要求较高的领域来说依然不够完美。
在这样的背景下，Flink 应运而生，接下来我们正式进入 Flink 的学习。

Flink 简介

Apache Flink 是为分布式、高性能、随时可用以及准确 的流处理应用程序打造的开源流处理框架，用于对无界和有界数据流进行有状态计算。Flink 最早起源于在 2010 ~ 2014 年，由 3 所地处柏林的大学和欧洲的一些其它大学共同进行研究的名为 Stratosphere 的项目。2014 年 4 月 Stratosphere 将其捐赠给 Apache 软件基 金会， 初始成员是 Stratosphere 系统的核心开发人员，2014 年 12 月，Flink 一跃成为 Apache 软件基金会的顶级项目。在 2015 年，阿里也加入到了 Flink 的开发工作中，并贡献了至少 150 万行代码。

Flink 一词在德语中有着“灵巧”、“快速”的意思，它的 logo 原型也是柏林常见的一种松鼠，以身材娇小、灵活著称，为该项目取这样的名字和选定这样的 logo 也正好符合 Flink 的特点和愿景。

注意，虽然我们说 Flink 是一个流处理框架，但是它同样可以进行批处理。因为在 Flink 的世界观里，批处理是流处理的一种特殊形式，这和 Spark 不同，在 Spark 中，流处理是通过大批量的微批处理实现的。

运行第一个 Flink 程序

接下来我们运行第一个 Flink 程序，感受一下它的魅力，从而对它有一个初步的印象。

搭建开发环境

本课程使用的是本地环境。后续实验不再提示。

首先我们需要在环境中搭建 Flink 运行环境，总共可以分为下面这几步：

• 安装 jdk 并配置环境变量 【jdk 1.8】
• 安装 scala 并配置环境变量 【Scala 2.11.12】
• 安装 maven 并修改中心仓库为阿里云地址
• 安装 IDEA 开发工具 【IDEA 2022.2.1或更新版】

我们的实验环境已经为大家安装了 jdk、scala、maven 和 IDEA，只需要在 IDEA 里配置使用即可。

双击桌面的 IDEA 程序，启动之后，点击 File -> New -> Project 创建一个新的 Maven 工程 FlinkLearning：

【或者New Project->Maven Archetype】

创建好之后点击左上角 File > Settings 中，将 Maven 的配置文件修改为 D:\programs\apahe-maven-3.6.3\conf\settings.xml，配置之后的 Maven 中心仓库为阿里云，加载依赖会快很多：

项目中点击File-》Project Structure -> Libraries-》加号-》添加Scala SDK-》选择所需要的scala版本-》ok进行下载

在工程 src/main 目录中创建 scala 文件夹，然后右键，选择 Mark Directory as，并将其标记为 Sources Root。

在 scala 目录里创建 com.vlab.wc 包，并分别创建 BatchWordCount 和 StreamWordCount 两个 Scala Object，分别代表 Flink 批处理和 Flink 流处理。

至此，我们的准备工作已经完成，接下来正式进入编码阶段。

Flink 批处理 WordCount

修改 pom.xml 文件中的代码为如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>pblh123.lh</groupId>
    <artifactId>FlinkLearning</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

<!--    配置国内 Maven 依赖库的镜像源镜-->
    <repositories>
        <repository>
            <id>aliyun</id>
            <url>https://maven.aliyun.com/repository/public</url>
        </repository>
    </repositories>
<!--配置插件的镜像源-->
    <pluginRepositories>
        <pluginRepository>
            <id>aliyun</id>
            <url>https://maven.aliyun.com/repository/public</url>
        </pluginRepository>
    </pluginRepositories>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>1.17.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>1.17.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>1.17.2</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <!-- 该插件用于将 Scala 代码编译成 class 文件 -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.4.6</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.6.0</version>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass></mainClass>
                        </manifest>
                    </archive>
                    <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>

</project>

修改完成后注意点击加载图标重新 Load。

BatchWordCount.scala 中的代码如下：

package com.vlab.wc

import org.apache.flink.api.scala._

/**
 * @projectName FlinkLearning
 * @package com.vlab.wc
 * @className com.vlab.wc.BatchWordCount
 * @description Flink Batch Word Count Example
 * @author pblh123
 * @date 2025/2/7 14:41
 * @version 1.17.2
 */
object BatchWordCount {
  def main(args: Array[String]): Unit = {
    // 判断输入参数数量是否正确
    if (args.length != 1) {
      System.err.println("Usage: BatchWordCount <input path>")
      System.exit(5)
    }
    // 获取输入路径
    val inputPath = args(0)
    // 创建执行环境
    val env = ExecutionEnvironment.getExecutionEnvironment
    // 读取输入数据
    val inputDS: DataSet[String] = env.readTextFile(inputPath)
    // 计算词频
    val wordCountDS: DataSet[(String, Int)] = inputDS
      .flatMap(_.split("\\s+")) // 扁平化，并且处理多个空格作为分隔符
      .map((_, 1)) // 转换为 (word, 1)
      .groupBy(0) // 按第一个字段（word）进行分组
      .sum(1) // 对第二个字段（计数）求和
    // 打印输出
    wordCountDS.print()
  }
}

在 datas/words.txt 路径下创建 words.txt 文件并在其中加入如下内容（注意每个单词之间使用空格分隔）：

hello world
hello flink
hello spark
hello java

右键选中 BatchWordCount.scala，点击 Run 运行，将会看到如下输出：

如果出现一些其他的报错和警告可以忽略。

(java,1)
(world,1)
(flink,1)
(hello,4)
(spark,1)

Flink 流处理 WordCount

在 StreamWordCount.scala 中加入如下代码：

package com.vlab.wc

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

/**
 * @projectName FlinkLearning  
 * @package com.vlab.wc  
 * @className com.vlab.wc.StreamWordCount  
 * @description ${description}  
 * @author pblh123
 * @date 2025/2/7 14:41
 * @version 1.0
 *
 */
    
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //  监控Socket数据
    val textDstream: DataStream[String] = env.socketTextStream("localhost", 9999)
    // 导入隐式转换
    import org.apache.flink.api.scala._
    // 计算逻辑
    val dataStream: DataStream[(String, Int)] = textDstream
      .flatMap(_.split(" "))
      .filter(_.nonEmpty)
      .map((_, 1))
      .keyBy(0)
      .sum(1)

    // 设置并行度
    dataStream.print().setParallelism(1)

    // 执行
    env.execute("Socket stream word count")
  }
}

打开终端，并输入 nc -l -p 9999，然后输入以下内容：

hello world
hello flink
hello spark
hello java

运行 StreamWordCount.scala 将会看到如下输出：

(hello,1)
(flink,1)
(hello,2)
(spark,1)
(java,1)
(hello,3)
...

至此，我们的第一个 Flink 实验就已经完成了。

实验总结

本节实验中，我们介绍了 Flink 出现的背景，并和 Storm、Spark Streaming 做了简单对比，然后在实验环境下安装了 idea 开发工具并运行了第一个 Flink 程序。至此，相信大家已经对 Flink 已经有了一个初步的认识。