2.Flink入门编程

环境准备
创建项目
- 创建工程
- 添加项目依赖
- 配置日志管理
编写代码
- 批处理
- - 运行程序
- 流处理
- - 读取文件
  - 运行程序
  - 读取文本流

环境准备

Flink 底层是以 Java 编写，但给开发提供了 Java 和 Scala API

使用 IntelliJ IDEA 作为开发工具，用 Maven 作为包管理工具

基于 Flink 1.13.0
基于 JDK 1.8

GitHub 代码托管 : https://github.com/CPU-Code/Hadoop

创建项目创建工程

打开 IntelliJ IDEA ，创建一个 Maven 工程

将这个 Maven 工程命名为 FlinkDemo , 选定这个 Maven 工程所在存储路径，并点击 Finish

添加项目依赖

在项目的 pom 文件中，增加标签设置属性，然后增加标签引入需要的依赖

Flink的架构中使用了 Akka 来实现底层的分布式通信，而 Akka 是用 Scala 开发

    
        1.13.0
        1.8
        2.12
        1.7.30
    

    
        
        
            org.apache.flink
            flink-java
            ${flink.version}
        
        
            org.apache.flink
            flink-streaming-java_${scala.binary.version}
            ${flink.version}
        
        
        
            org.apache.flink
            flink-clients_${scala.binary.version}
            ${flink.version}
        

        
        
            org.slf4j
            slf4j-api
            ${slf4j.version}
        
        
            org.slf4j
            slf4j-log4j12
            ${slf4j.version}
        
        
            org.apache.logging.log4j
            log4j-to-slf4j
            2.14.0

配置日志管理

在目录 src/main/resources 下添加文件: log4j.properties

内容配置 :

log4j.rootLogger=error, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n

编写代码

需求 : 统计一段文字中，每个单词出现的频次

源码位于 src/main/java 目录下。首先新建一个包，命名为 com.cpucode.wc

批处理

批处理 : 输入是收集好的数据集。将要统计的文字，写入一个文本文档，然后读取这个文件处理数据

在目录 src/main/resources 下新建一个 input 文件夹，并在下面创建文本文件 words.txt

在 words.txt 中输入一些文字 :

hello world
hello flink
hello java

创建建 Java 类 BatchWordCount，在静态 main 方法中编写测试代码

进行单词频次统计的基本思路是：

先逐行读入文件数据，然后将每一行文字拆分成单词
接着按照单词分组，统计每组数据的个数，就是对应单词的频次

具体代码实现 :

package com.cpucode.wc;

// 在引入包时，有 Java 和 Scala ，注意选用 Java 的包
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.util.Collector;


public class BatchWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境 , 获取执行环境对象，也就是运行时上下文环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)
        DataSource lineDS = env.readTextFile("FlinkDemo/src/main/resources/input/words.txt");

        // 3. 转换数据格式 flatmap 方法可以对一行文字进行分词转换
        FlatMapOperator> wordAndOne = lineDS.flatMap((String line, Collector> out) -> {
            // 每一行文字拆分成单词
            String[] words = line.split(" ");
            for (String word : words) {
                // 转换成(word,count)形式的二元组
                out.collect(Tuple2.of(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));
        //当 Lambda 表达式使用 Java 泛型的时候, 由于泛型擦除的存在, 需要显示的声明类型信息


        // 4. 按照 word 进行分组 , 采用位置索引或属性名称进行分组
        UnsortedGrouping> wordAndOneUG = wordAndOne.groupBy(0);

        // 5. 分组内聚合统计 , 指定聚合字段的位置索引或属性名称
        AggregateOperator> sum = wordAndOneUG.sum(1);

        // 6. 打印结果
        sum.print();
    }
}

运行程序

将文档中的所有单词的频次，全部统计出来，以二元组的形式在控制台打印输出

上面的实现方式，是基于 DataSet API ( 软弃用 ) , 从 Flink 1.12 开始，推荐使用 DataStream API，流批统一的处理架构，在提交任务时将执行模式设为 BATCH 来进行批处理：

bin/flink run -Dexecution.runtime-mode=BATCH BatchWordCount.jar

流处理

用 DataSet API 实现批处理
用 DataStream API 实现流处理

Flink : 流才是整个处理逻辑的底层核心，所以DataStream API实现了流批统一，可以直接处理批处理和流处理的所有场景

读取文件

读取文档 words.txt 中的数据，并统计每个单词出现的频次

在 com.cpucode.wc 包下新建 Java 类 BoundedStreamWordCount ，在静态 main 方法中编写测试代码

具体代码实现 :

package com.cpucode.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;


public class BoundedStreamWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建流式执行环境 , 流处理
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取文件
        DataStreamSource lineDSS = env.readTextFile("FlinkDemo/src/main/resources/input/words.txt");

        // 3. 转换数据格式
        SingleOutputStreamOperator> wordAndOne = lineDSS.flatMap((String line, Collector word) -> {
            Arrays.stream(line.split(" ")).forEach(word::collect);
        }).returns(Types.STRING)
                .map(word -> Tuple2.of(word, 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG));

        // 4. 分组 , 匿名函数作为键选择器
        KeyedStream, String> wordAndOneKS = wordAndOne.keyBy(t -> t.f0);

        // 5. 求和
        SingleOutputStreamOperator> result = wordAndOneKS.sum(1);

        // 6. 打印
        result.print();

        // 7. 开始执行任务
        env.execute();
    }
}

运行程序

与批处理的结果是完全不同

批处理针对每个单词，只会输出一个最终的统计个数
而流处理的打印结果中，单词每出现一次，都会有一个频次统计数据输出。

这就是流处理的特点，数据逐个处理，每来一条数据就会处理输出一次

显示的编号为 1~16，是 CPU 是 16 核，所以默认模拟的并行线程是 16 个。不同的运行环境，得到的结果会是不同的

读取文本流

在实际的生产环境中，真正的数据流是无界的，只有开始没有结束

为了模拟这种场景，就不读取文件来获取数据了，而是监听数据发送端主机的指定端口，统计发送来的文本数据中出现过的单词的个数

新建一个 Java 类 StreamWordCount，将 BoundedStreamWordCount 代码中读取文件数据的 readTextFile 方法，替换成读取 socket 文本流的方法 socketTextStream

具体代码实现 :

package com.cpucode.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.lang.reflect.Array;
import java.util.Arrays;


public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取文本流 , 发送端主机名和端口号
        DataStreamSource lineDSS = env.socketTextStream("cpu101", 7777);

        // 3. 转换数据格式
        SingleOutputStreamOperator> wordAndOne = lineDSS.flatMap((String line, Collector words) -> {
            Arrays.stream(line.split(" ")).forEach(words::collect);
        }).returns(Types.STRING)
                .map(word -> Tuple2.of(word, 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG));

        // 4. 分组
        KeyedStream, String> wordAndOneKS = wordAndOne.keyBy(t -> t.f0);

        // 5. 求和
        SingleOutputStreamOperator> result = wordAndOneKS.sum(1);

        // 6. 打印
        result.print();

        // 7. 执行
        env.execute();
    }
}

在 Linux 环境的主机 cpu101 上，发送数据进行测试

安装 netcat 工具

yum install -y nc

nc -lk 7777

运行程序

程序启动后没有任何输出也不会退出。 Flink 的流处理是事件驱动的，程序会一直处于监听状态，只有接收到数据才会执行任务、输出统计结

从 cpu101 发送数据：

看到控制台输出结果 :

2.Flink入门编程

Java相关栏目本月热门文章