实时计算框架：Flink集群搭建与运行机制

知了一笑 2023-03-28 原文

一、Flink概述

1、基础简介

Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。Flink被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。主要特性包括：批流一体化、精密的状态管理、事件时间支持以及精确一次的状态一致性保障等。Flink不仅可以运行在包括YARN、Mesos、Kubernetes在内的多种资源管理框架上，还支持在裸机集群上独立部署。在启用高可用选项的情况下，它不存在单点失效问题。

这里要说明两个概念：

边界：无边界和有边界数据流，可以理解为数据的聚合策略或者条件；
状态：即执行顺序上是否存在依赖关系，即下次执行是否依赖上次结果；

2、应用场景

Data Driven

事件驱动型应用无须查询远程数据库，本地数据访问使得它具有更高的吞吐和更低的延迟，以反欺诈案例来看，DataDriven把处理的规则模型写到DatastreamAPI中，然后将整个逻辑抽象到Flink引擎，当事件或者数据流入就会触发相应的规则模型，一旦触发规则中的条件后，DataDriven会快速处理并对业务应用进行通知。

Data Analytics

和批量分析相比，由于流式分析省掉了周期性的数据导入和查询过程，因此从事件中获取指标的延迟更低。不仅如此，批量查询必须处理那些由定期导入和输入有界性导致的人工数据边界，而流式查询则无须考虑该问题，Flink为持续流式分析和批量分析都提供了良好的支持，实时处理分析数据，应用较多的场景如实时大屏、实时报表。

Data Pipeline

与周期性的ETL作业任务相比，持续数据管道可以明显降低将数据移动到目的端的延迟，例如基于上游的StreamETL进行实时清洗或扩展数据，可以在下游构建实时数仓，确保数据查询的时效性，形成高时效的数据查询链路，这种场景在媒体流的推荐或者搜索引擎中十分常见。

二、环境部署

1、安装包管理

[root@hop01 opt]# tar -zxvf flink-1.7.0-bin-hadoop27-scala_2.11.tgz
[root@hop02 opt]# mv flink-1.7.0 flink1.7

2、集群配置

管理节点

[root@hop01 opt]# cd /opt/flink1.7/conf
[root@hop01 conf]# vim flink-conf.yaml

jobmanager.rpc.address: hop01

分布节点

[root@hop01 conf]# vim slaves

hop02
hop03

两个配置同步到所有集群节点下面。

3、启动与停止

/opt/flink1.7/bin/start-cluster.sh
/opt/flink1.7/bin/stop-cluster.sh

启动日志：

[root@hop01 conf]# /opt/flink1.7/bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host hop01.
Starting taskexecutor daemon on host hop02.
Starting taskexecutor daemon on host hop03.

4、Web界面

访问：http://hop01:8081/

三、开发入门案例

1、数据脚本

分发一个数据脚本到各个节点：

/var/flink/test/word.txt

2、引入基础依赖

这里基于Java写的基础案例。

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.7.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>1.7.0</version>
    </dependency>
</dependencies>

3、读取文件数据

这里直接读取文件中的数据，经过程序流程分析出每个单词出现的次数。

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 读取文件数据
        readFile () ;
    }

    public static void readFile () throws Exception {
        // 1、执行环境创建
        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();

        // 2、读取数据文件
        String filePath = "/var/flink/test/word.txt" ;
        DataSet<String> inputFile = environment.readTextFile(filePath);

        // 3、分组并求和
        DataSet<Tuple2<String, Integer>> wordDataSet = inputFile.flatMap(new WordFlatMapFunction(
        )).groupBy(0).sum(1);

        // 4、打印处理结果
        wordDataSet.print();
    }

    // 数据读取个切割方式
    static class WordFlatMapFunction implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String input, Collector<Tuple2<String, Integer>> collector){
            String[] wordArr = input.split(",");
            for (String word : wordArr) {
                collector.collect(new Tuple2<>(word, 1));
            }
        }
    }
}

4、读取端口数据

在hop01服务上创建一个端口，并模拟一些数据发送到该端口：

[root@hop01 ~]# nc -lk 5566
c++,java

通过Flink程序读取并分析该端口的数据内容：

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 读取端口数据
        readPort ();
    }

    public static void readPort () throws Exception {
        // 1、执行环境创建
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2、读取Socket数据端口
        DataStreamSource<String> inputStream = environment.socketTextStream("hop01", 5566);

        // 3、数据读取个切割方式
        SingleOutputStreamOperator<Tuple2<String, Integer>> resultDataStream = inputStream.flatMap(
                new FlatMapFunction<String, Tuple2<String, Integer>>()
        {
            @Override
            public void flatMap(String input, Collector<Tuple2<String, Integer>> collector) {
                String[] wordArr = input.split(",");
                for (String word : wordArr) {
                    collector.collect(new Tuple2<>(word, 1));
                }
            }
        }).keyBy(0).sum(1);

        // 4、打印分析结果
        resultDataStream.print();

        // 5、环境启动
        environment.execute();
    }
}

四、运行机制

FlinkClient

客户端用来准备和发送数据流到JobManager节点，之后根据具体需求，客户端可以直接断开连接，或者维持连接状态等待任务处理结果。

JobManager

在Flink集群中，会启动一个JobManger节点和至少一个TaskManager节点，JobManager收到客户端提交的任务后，JobManager会把任务协调下发到具体的TaskManager节点去执行，TaskManager节点将心跳和处理信息发送给JobManager。

TaskManager

任务槽（slot）是TaskManager中最小的资源调度单位，在启动的时候就设置好了槽位数，每个槽位能启动一个Task，接收JobManager节点部署的任务，并进行具体的分析处理。

$$End$$

Gitee主页：https://gitee.com/cicadasmile/butte-java-note

Flink 实时 code String strong Spark 大数据实时计算流式计算Spark

有关实时计算框架：Flink集群搭建与运行机制的更多相关文章

ruby - 如何从 ruby 中的字符串运行任意对象方法？ - 2
总的来说，我对ruby还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用
ruby - 如何每月在 Heroku 运行一次 Scheduler 插件？ - 2
在选择我想要运行操作的频率时，唯一的选项是“每天”、“每小时”和“每10分钟”。谢谢!我想为我的Rails3.1应用程序运行调度程序。最佳答案这不是一个优雅的解决方案，但您可以安排它每天运行，并在实际开始工作之前检查日期是否为当月的第一天。关于ruby-如何每月在Heroku运行一次Scheduler插件？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.com/questions/8692687/
ruby-on-rails - 如何在 ruby 中使用两个参数异步运行 exe？ - 2
exe应该在我打开页面时运行。异步进程需要运行。有什么方法可以在ruby中使用两个参数异步运行exe吗？我已经尝试过ruby命令-system()、exec()但它正在等待过程完成。我需要用参数启动exe，无需等待进程完成是否有任何rubygems会支持我的问题？最佳答案您可以使用Process.spawn和Process.wait2:pid=Process.spawn'your.exe','--option'#Later...pid,status=Process.wait2pid您的程序将作为解释器的子进程执行。除
ruby - 无法运行 Rails 2.x 应用程序 - 2
我尝试运行2.x应用程序。我使用rvm并为此应用程序设置其他版本的ruby:$rvmuseree-1.8.7-head我尝试运行服务器，然后出现很多错误:$script/serverNOTE:Gem.source_indexisdeprecated,useSpecification.Itwillberemovedonorafter2011-11-01.Gem.source_indexcalledfrom/Users/serg/rails_projects_terminal/work_proj/spohelp/config/../vendor/rails/railties/lib/r
ruby-on-rails - 使用一系列等级计算字母等级 - 2
这里是Ruby新手。完成一些练习后碰壁了。练习:计算一系列成绩的字母等级创建一个方法get_grade来接受测试分数数组。数组中的每个分数应介于0和100之间，其中100是最大分数。计算平均分并将字母等级作为字符串返回，即“A”、“B”、“C”、“D”、“E”或“F”。我一直返回错误:avg.rb:1:syntaxerror,unexpectedtLBRACK,expecting')'defget_grade([100,90,80])^avg.rb:1:syntaxerror,unexpected')',expecting$end这是我目前所拥有的。我想坚持使用下面的方法或.join，
ruby - Sinatra:运行 rspec 测试时记录噪音 - 2
Sinatra新手；我正在运行一些rspec测试，但在日志中收到了一堆不需要的噪音。如何消除日志中过多的噪音？我仔细检查了环境是否设置为:test，这意味着记录器级别应设置为WARN而不是DEBUG。spec_helper:require"./app"require"sinatra"require"rspec"require"rack/test"require"database_cleaner"require"factory_girl"set:environment,:testFactoryGirl.definition_file_paths=%w{./factories./test/
ruby-on-rails - 无法让 rspec、spork 和调试器正常运行 - 2
GivenIamadumbprogrammerandIamusingrspecandIamusingsporkandIwanttodebug...mmm...let'ssaaay,aspecforPhone.那么，我应该把“require'ruby-debug'”行放在哪里，以便在phone_spec.rb的特定点停止处理？(我所要求的只是一个大而粗的箭头，即使是一个有挑战性的程序员也能看到:-3)我已经尝试了很多位置，除非我没有正确测试它们，否则会发生一些奇怪的事情:在spec_helper.rb中的以下位置:require'rubygems'require'spork'
ruby-on-rails - before_filter 运行多个方法 - 2
是否有可能:before_filter:authenticate_user!||:authenticate_admin! 最佳答案 before_filter:do_authenticationdefdo_authenticationauthenticate_user!||authenticate_admin!end 关于ruby-on-rails-before_filter运行多个方法，我们在StackOverflow上找到一个类似的问题： https://
Vscode+Cmake配置并运行opencv环境(Windows和Ubuntu大同小异) - 2
之前在培训新生的时候，windows环境下配置opencv环境一直教的都是网上主流的vsstudio配置属性表，但是这个似乎对新生来说难度略高(虽然个人觉得完全是他们自己的问题)，加之暑假之后对cmake实在是爱不释手，且这样配置确实十分简单(其实都不需要配置)，故斗胆妄言vscode下配置CV之法。其实极为简单，图比较多所以很长。如果你看此文还配不好，你应该思考一下是不是自己的问题。闲话少说，直接开始。0.CMkae简介有的人到大二了都不知道cmake是什么，我不说是谁。CMake是一个开源免费并且跨平台的构建工具，可以用简单的语句来描述所有平台的编译过程。它能够根据当前所在平台输出对应的m
计算机毕业设计ssm+vue基本微信小程序的小学生兴趣延时班预约小程序 - 2
项目介绍随着我国经济迅速发展,人们对手机的需求越来越大,各种手机软件也都在被广泛应用,但是对于手机进行数据信息管理,对于手机的各种软件也是备受用户的喜爱小学生兴趣延时班预约小程序的设计与开发被用户普遍使用,为方便用户能够可以随时进行小学生兴趣延时班预约小程序的设计与开发的数据信息管理,特开发了小程序的设计与开发的管理系统。小学生兴趣延时班预约小程序的设计与开发的开发利用现有的成熟技术参考,以源代码为模板,分析功能调整与小学生兴趣延时班预约小程序的设计与开发的实际需求相结合,讨论了小学生兴趣延时班预约小程序的设计与开发的使用。开发环境开发说明：前端使用微信微信小程序开发工具：后端使用ssm：VU