Flink 从入门到实战

🕗 发布于 2024-11-26 06:10 flink 大数据

Flink中的批和流

批处理的特点是有界、持久、大量，非常适合需要访问全部记录才能完成的计算工作，一般用于离线统计。

流处理的特点是无界、实时, 无需针对整个数据集执行操作，而是对通过系统传输的每个数据项执行操作，一般用于实时统计。

一个无界流可以分解为多个有界流

性能 Flink > Spark > Hadoop

Flink的四种安装模式：

Local
Standalone
standaloneHA
Yarn

flink在使用input、output执行测试文件WordCount.jar 的时候，报出找不到文件的错误（但是文件路径存在），原因是：

因为我们的flink是task节点在执行任务的，task在三台机器上都有分布，我们当前文件只在一台服务器中，所以当其他task运行的时候，就会报出找不到文件的错误，将此文件分发到每台服务器中就不会出现这个错误。（我们以后在使用flink的时候，数据都是存放在hdfs上（一式三份），就不存在找不到文件的错误）

Flink-WordCount案例：

第一版代码

这一版代码比较简单，看代码就可以看懂

package com.bigdata;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
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.util.Collector;

public class _01WorkCount {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> dataStream01 = env.fromElements("spark flink kafka", "spark sqoop flink", "kafka hadoop flink");

        // 首先先对字符串进行切割，形成一个新的数组
        SingleOutputStreamOperator<String> flatMapStream = dataStream01.flatMap(new FlatMapFunction<String, String>() {

            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {

                String[] arr = line.split(" ");
                for (String word : arr) {
                    collector.collect(word);
                }
            }
        });

        // 将切割好的字符串形成 （word，1）的二元组的形式
        SingleOutputStreamOperator<Tuple2<String, Integer>> map = flatMapStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                return Tuple2.of(word, 1);

            }
        });

        // 聚合
        DataStream<Tuple2<String, Integer>> sumResult = map.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> tuple2) throws Exception {
                return tuple2.f0;
            }
            // 此处的1 指的是元组的第二个元素，进行相加的意思
        }).sum(1);

        sumResult.print();
        env.execute();

    }
}

第二版代码：简化了第一版的书写形式

第一版中 SingleOutputStreamOperator、DataStreamSource的父类其实都是DataStream，所以可以连着写下来

package com.bigdata;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
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.util.Collector;

public class _02WorkCount {

    /**
     *
     *  简化版案例
     * @param args
     * @throws Exception
     */

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.fromElements("spark flink kafka", "spark sqoop flink", "kafka hadoop flink")
                .flatMap(new FlatMapFunction<String, String>() {

            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {

                String[] arr = line.split(" ");
                for (String word : arr) {
                    collector.collect(word);
                }
            }
        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                return Tuple2.of(word, 1);

            }
        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> tuple2) throws Exception {
                return tuple2.f0;
            }
            // 此处的1 指的是元组的第二个元素，进行相加的意思
        }).sum(1).print();
        env.execute();

    }
}

第三版，使用lambda表达式，更加简化的书写

不过在使用lambda的时候，需要在后面指定一个方法的返回值，要不然会报错

package com.bigdata;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
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.util.Collector;

public class _03WorkCount_lambda {

    /**
     * lambda 表达式简化版
     * @param args
     * @throws Exception
     */

    public static void main(String[] args) throws Exception {

        // 使用lambda简化的时候，需要指定返回值类型，不指定的话会报错

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.fromElements("spark flink kafka", "spark sqoop flink", "kafka hadoop flink")
                .flatMap((String line, Collector<String> collector) -> {

                String[] arr = line.split(" ");
                for (String word : arr) {
                    collector.collect(word);
                }
            }).returns(Types.STRING).map((String word) ->  Tuple2.of(word, 1)

            ).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy((Tuple2<String, Integer> tuple2) ->  tuple2.f0).sum(1).print();
            // 此处的1 指的是元组的第二个元素，进行相加的意思
        env.execute();

    }
}

复习lambda表达式：

lambda可以用来简化匿名内部类的书写
lambda只能简化函数式接口（有且仅有一个方法的接口）的匿名内部类的书写

省略规则：

只拿小括号里面的加上 -> 指向大括号
只有一个参数的时候，参数类型可以省略
如果方法体中的代码只有一行，大括号和return等都可以省略（但是需要同时省略）

没省略之前的 (第一版)

省略后（第三版）

第四版，自定义输入与输出的路径地址

可以打包到集群中使用，使用的时候在jar包的后面跟上input路径以及output路径即可

package com.bigdata;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
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.util.Collector;

public class _04WorkCount_zidingyipass {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 这个是 自动 ，根据流的性质，决定是批处理还是流处理
        //env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 批处理流， 一口气把数据算出来
        // env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        // 流处理，默认是这个  可以通过打印批和流的处理结果，体会流和批的含义
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        // 将任务的并行度设置为2
        // env.setParallelism(2);

        // 通过args传参
        DataStreamSource<String> dataStream01 = null;
        if (args.length == 0){
            dataStream01 = env.fromElements("spark flink kafka", "spark sqoop flink", "kafka hadoop flink");
        }else {
            String input = args[0];
            dataStream01 = env.readTextFile(input);
        }


        // 首先先对字符串进行切割，形成一个新的数组
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumResult = dataStream01
                .flatMap((String line, Collector<String> collector) -> {

                String[] arr = line.split(" ");
                for (String word : arr) {
                    collector.collect(word);
                }

        }).map((String word) -> Tuple2.of(word, 1)


        ).keyBy((Tuple2<String, Integer> tuple2) -> tuple2.f0

         // 此处的1 指的是元组的第二个元素，进行相加的意思
        ).sum(1);

        if (args.length == 0){
            sumResult.print();
        }else {
            String output = args[1];
            sumResult.writeAsText(output, FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        }

        env.execute();

    }
}

打包后执行结果如下：

第五版，采用和flink中相同的书写方式即带（--input 以及 --output）

也可以打包到集群中使用，使用的时候在jar包的后面跟上 --input +路径以及 -output + 路径即可

package com.bigdata;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.core.fs.FileSystem;
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.util.Collector;

public class _05WorkCount_zidingyipass_input {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 这个是 自动 ，根据流的性质，决定是批处理还是流处理
        //env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 批处理流， 一口气把数据算出来
        // env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        // 流处理，默认是这个  可以通过打印批和流的处理结果，体会流和批的含义
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        // 将任务的并行度设置为2
        // env.setParallelism(2);
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String input = "";
        String output = "";
        if (parameterTool.has("output") && parameterTool.has("input")) {
            input = parameterTool.get("input");
            output = parameterTool.get("output");
        } else {
            output = "hdfs://bigdata01:9820/home/wordcount/output";
        }

        // 通过args传参
        DataStreamSource<String> dataStream01 = null;
        if (args.length == 0){
            dataStream01 = env.fromElements("spark flink kafka", "spark sqoop flink", "kafka hadoop flink");
        }else {
            dataStream01 = env.readTextFile(input);
        }
        // 首先先对字符串进行切割，形成一个新的数组
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumResult = dataStream01
                .flatMap((String line, Collector<String> collector) -> {
                
                String[] arr = line.split(" ");
                for (String word : arr) {
                    collector.collect(word);
                }
        }).returns(Types.STRING).map((String word) -> Tuple2.of(word, 1)

        ).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy((Tuple2<String, Integer> tuple2) -> tuple2.f0

         // 此处的1 指的是元组的第二个元素，进行相加的意思
        ).sum(1);

        if (args.length == 0){
            sumResult.print();
        }else {
            sumResult.writeAsText(output, FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        }
        env.execute();
    }
}

原文地址：https://blog.csdn.net/weixin_53709379/article/details/144039237

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：【Java 学习】构造器、static静态变量、static静态方法、static构造器、
下一篇：springboot获取配置文件中的值

nvm 安装某个版本的node，缺少npm包
1、根据地址下载对应文件 https://github.com/npm/cli/archive/v6.14.16.zip。2、解压文件到v12.22.12 的node_modules/npm 目录下。
阅读更多2024-11-27
ffmpeg RTP PS推流
流媒体开发
阅读更多2024-11-27
JavaWeb——SpringBoot原理
properties > yml(推荐) > yaml命令行参数 > Java系统属性 > 配置文件自定义的类，使用@component及其衍生注解第三方的，使用@Bean注解
阅读更多2024-11-27
【系统架构设计师】真题论文: 论软件系统架构评估（包括解题思路和素材）
软件系统架构是软件质量的基石。架构分析与权衡：根据收集的场景和构建的属性模型，对架构进行深入分析，评估架构在满足不同质量属性方面的优势和劣势，并进行权衡。例如，在考虑提高系统的性能时，可能需要增加硬件
阅读更多2024-11-27
CodeIgniter如何手动将模型连接到数据库
在CodeIgniter中，模型通常是自动与数据库连接的，因为模型类（CI_Model）已经内置了对数据库操作的支持。但是，如果你需要手动指定数据库连接或者进行一些特殊的数据库配置，你可以通过几种方式
阅读更多2024-11-27
uname -m（machine）命令用于显示当前系统的机器硬件架构（Unix Name）
这意味着您的 Mac Mini 使用的是 Apple 的 M1 或更新的芯片，这些芯片基于 ARM 架构。希望这些信息对您有帮助！如果您有其他问题或需要进一步的帮助，请随时告诉我。命令可以查看系统的详
阅读更多2024-11-27
微服务篇-深入了解使用 RestTemplate 远程调用、Nacos 注册中心基本原理与使用、OpenFeign 的基本使用
比如刚才通过 Swagger 测试商品查询接口，而这种查询就是通过 http 请求的方式来完成的，不仅仅可以实现远程查询，还可以实现新增、删除等各种远程请求。项目的架构设计、开发模式都非常简单。在微服
阅读更多2024-11-27
[自动化测试：实践01]：2：（4-1 ）元素定位（selenium）在实际场景中的应用2
熟悉元素定位的基本用法。结合ActionChains实现网页上的鼠标自动操作测试。
阅读更多2024-11-27
【小白学机器学习36】关于独立概率，联合概率，交叉概率，交叉概率和，总概率等概念辨析的例子
关于独立概率，联合概率，交叉概率，交叉概率和，总概率类型含义计算联合概率两个独立事件一起发生的概率两个事件概率相乘边缘概率同1行 /同1列的所有联合概率相加的总和两个联合概率相加条件概率一定已知条件下
阅读更多2024-11-27
mysql覆盖索引&回表查询
就是查询的字段是索引里的。打个比方，有个user表，字段为id、name、gender，id是主键，有聚簇索引，name是非聚簇索引。第一条语句符合覆盖索引，因为查询的字段id和name是是name索
阅读更多2024-11-27