大数据-108 Flink 快速应用案例重回Hello WordCount！方案1批数据方案2流数据

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（正在更新！）

章节内容

上节完成了如下的内容：

Flink 基本介绍
基本概述适用场景框架特点核心组成生态发展处理模型组件架构

在这里插入图片描述

再次回到最初的起点，Hello Word Count！

Flink 流处理 (Stream Processing)

定义

流处理是指对持续不断的数据流进行实时处理。Flink 的流处理模式非常适合处理持续产生的数据，例如来自传感器、日志记录系统或金融交易的数据流。

核心概念

无界数据流：流处理通常处理无界数据流，即数据流没有明确的结束点，持续不断地产生。
事件时间：Flink 支持基于事件时间的处理，能够处理乱序和延迟数据，使得处理更加精确。事件时间指的是数据在其产生源头的时间。
窗口操作：在流处理过程中，通常需要将数据按时间窗口（如滑动窗口、滚动窗口、会话窗口）进行分组，以便执行聚合或其他操作。
状态管理：Flink 支持有状态的流处理，这意味着处理每条数据时，可以记住之前的数据状态。例如，在流中计算一个累积的总和或频率。

Flink 批处理 (Batch Processing)

定义

批处理是指对静态的、有界的数据集进行处理。这种处理通常用于一次性的大规模数据分析，如定期的业务报告生成、数据转换和加载任务。

核心概念

有界数据集：批处理通常处理有界数据集，即数据集是固定大小的，有明确的开始和结束点。
任务并行化：在批处理模式下，Flink 会将数据集划分为多个子任务，并行执行这些任务，以加快处理速度。
DataSet API：Flink 的 DataSet API 提供了一组高层次的操作符，用于对批数据集执行各种操作，如映射（map）、过滤（filter）、联接（join）和聚合（aggregate）。

单词统计（批数据）

需求说明

统计一个文件中各个单词出现的次数，把统计结果输出到文件

读取数据源
处理数据源
将读取到的数据源文件中的每一行根据空格切分
将切分好的每个单词拼接1
根据单词聚合（将相同的单词放到一起）
累加相同的单词（单词后面的1进行累加）
保存处理结果

导入依赖

<?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>org.example</groupId>
    <artifactId>flink-test</artifactId>
    <version>1.0-SNAPSHOT</version>

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

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.11.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.11.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.11.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>1.11.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>1.11.1</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

</project>

编写代码

package icu.wzk;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.util.Collector;


public class WordCount {

    public static void main(String[] args) throws Exception {
        String inPath = "word-count/word-count.txt";
        String outPath = "word-count/word-count-result.csv";
        // 获取Flink批处理执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 读取文件中的内容
        DataSet<String> text = env.readTextFile(inPath);
        // 对数据进行处理
        DataSet<Tuple2<String, Integer>> dataSet = text
                .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                        for (String word : line.split(" ")) {
                            collector.collect(new Tuple2<>(word, 1));
                        }
                    }
                })
                .groupBy(0)
                .sum(1);
        dataSet
                .writeAsCsv(outPath, "\n", " ", FileSystem.WriteMode.OVERWRITE)
                .setParallelism(1);
        // 触发执行程序
        env.execute("Word Count");
    }

}

测试数据

Stateful Computations over Data Streams
Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams.
Flink has been designed to run in all common cluster environments, perform computations at in-memory speed and at any scale.
Correctness guarantees
Exactly-once state consistency
Event-time processing
Sophisticated late data handling
SQL on Stream & Batch Data
DataStream API & DataSet API
ProcessFunction (Time & State)
Flexible deployment
High-availability setup
Savepoints

运行测试

在这里插入图片描述

结果数据

查看 word-count/word-count-result.csv 打开即可看到以下内容：

Stateful 1
any 1
common 1
computations 2
on 1
setup 1
state 1
streams. 1
unbounded 1
& 3
Data 2
DataStream 1
High-availability 1
for 1
perform 1
run 1
to 1
Event-time 1
Flexible 1
Sophisticated 1
framework 1
is 1
scale. 1
Exactly-once 1
ProcessFunction 1
Stream 1
a 1
been 1
handling 1
in 1
late 1
processing 2
Batch 1
DataSet 1
at 2
bounded 1
consistency 1
deployment 1
distributed 1
engine 1
has 1
API 2
Apache 1
Flink 2
SQL 1
Streams 1
all 1
designed 1
over 2
Computations 1
Savepoints 1
and 3
data 2
environments, 1
in-memory 1
speed 1
stateful 1
(Time 1
Correctness 1
State) 1
cluster 1
guarantees 1

单词统计（流数据）

需求说明

Socket模拟实时发送单词，使用Flink实时接收数据，对指定时间窗口内（如5秒）的数据进行聚合统计，每隔1秒汇总计算一次，并且把时间窗口内计算结果打印出来。

编写代码

Server部分

编写一个Socket服务，提供一定的数据流。

package icu.wzk;


import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;

public class WordCountServer {

    public static void main(String[] args) throws IOException, InterruptedException {
        String ip = "localhost";
        int port = 9999;
        Random random = new Random();
        ServerSocket serverSocket = new ServerSocket();
        InetSocketAddress address = new InetSocketAddress(ip, port);
        serverSocket.bind(address);
        Socket socket = serverSocket.accept();
        OutputStream outputStream = socket.getOutputStream();
        PrintWriter writer = new PrintWriter(outputStream, true);
        for (int i = 0; i < 1000; i ++) {
            int number = random.nextInt(100);
            System.out.println(number);
            writer.println(number);
            Thread.sleep((random.nextInt(900) + 100));
        }
        socket.close();
        serverSocket.close();
    }

}

Flink部分

连接到上述的Server部分

package icu.wzk;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;


public class WordCount2 {

    public static void main(String[] args) throws Exception {
        String ip = "localhost";
        int port = 9999;

        // 获取 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 获取 Socket 输入数据
        DataStreamSource<String> textStream = env.socketTextStream(ip, port, "\n");
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordCount = textStream
                .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        String[] splits = value.split("\\s");
                        for (String word : splits) {
                            out.collect(new Tuple2<>(word, 1));
                        }
                    }
                });
        SingleOutputStreamOperator<Tuple2<String, Integer>> word = wordCount
                .keyBy(new KeySelector<Tuple2<String, Integer>, Object>() {
                    @Override
                    public Object getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0;
                    }
                })
                .timeWindow(Time.seconds(5), Time.seconds(1))
                .sum(1);

        // 输出并运行
        word.print();
        env.execute("Word Count");
    }

}

观察结果

Server部分

Flink部分

3> (35,1)
4> (18,1)
3> (35,1)
5> (84,1)
4> (18,1)
6> (72,1)
3> (35,1)
5> (84,1)
5> (24,1)
3> (35,1)
6> (72,1)
4> (18,1)
7> (51,1)
5> (24,1)
5> (84,1)
4> (15,1)
6> (72,1)
7> (51,1)
3> (35,1)
4> (15,1)
4> (18,1)