Flink 环境的搭建、独立集群、Flink on Yarn、访问Flink web界面、Flink提 ...

一、Flink简介介绍

Flink：框架和（分布式）引擎，对（有界和无界）数据流举行有状态（即存储中间效果）盘算。

• 无界数据：有界说的开始，没有竣事，必须持续处理，即摄取到数据立即处理
  
• 有界数据：有界说的开始，也有竣事，摄取到所有数据后再盘算，可以被排序，即无需有序摄取，通常称为批处理。
  
• 存储位置：
  

              内存：速度快，可靠性差
              分布式体系：速度慢，可靠性强
 
DataStream 批流统一处理
       DataStream 批流统一处理，数据流的统一处理接口。流处理中有两种不同的流：

• 批处理，是将其数据看成有界(有界说开始，有界说竣事)流处理，例如文本文件数据
  
• 流处理，是将其数据看成无界(无界说开始，无界说竣事)流处理，例如实时数据
  

二、Flink集群的架构

三、Flink三种运行方式（与spark相似）

1、local 本地测试
2、Standallone Cluster 独立集群（做实时盘算，不必要hadoop，该独立集群可能用的上）
3、Flink on Yarn 推荐
四、底子环境搭建

1、服务器资源配置预备

     在文件/etc/hosts追加内容，配置映射地址，倒霉用假造映射直接利用ip也是可以

1. 192.168.56.128 hadoop001
2. 192.168.56.129 hadoop002
3. 192.168.56.130 hadoop003

复制代码

    实际的ip根据服务器资源修改
2、Java环境配置

    将下载好的Java举行安装，在文件 /root/.bash_profile或/etc/profile后追加内容。

1. export JAVA_HOME=/usr/local/java
2. export PATH=$PATH:$JAVA_HOME/bin
3. export CLASSPATH=.:$JAVA_HIOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

复制代码

   摆设前提，最Flink新版本1.17要求java 11以上版本，1.12还可以利用java 8版本。根据版本自行选择对应版本JDK。；
   注意：/root/.bash_profile体现体系环境（体系下所有用户生效）
               /etc/profile体现用户环境（只有当前用户生效）
3、使环境变量生效

1. source /root/.bash_profile

复制代码

4、验证是否乐成

1. java -version

复制代码

5、Flink上传、解压、配置环境变量

1. #进入压缩包所在目录
2. cd /usr/local/
3. #解压
4. tar -zxvf /usr/local/flink-1.17.0-bin-scala_2.12.tgz
6. #重命名
7. mv flink-1.17.0 flink
9. #配置环境变量
10. vi /root/.bash_profile
12. #添加
13. export FLINK_HOME=/usr/local/flink
15. export PATH=$PATH:$FLINK_HOME/bin
17. #刷新
18. source /root/.bash_profile

复制代码

6、防火墙设置

centos的防火墙要关闭，并确认是否关闭乐成。
会用到的命令如下，
a. 检察防火墙状态：firewall-cmd --state
假如是not running状态，分析没有启动防火墙
b. 关闭防火墙：systemctl stop firewalld.service
c. 设置开机禁启：systemctl disable firewalld.service
========================
以下为常用防火墙命令
检察防火墙状态：firewall-cmd --state
检察防火墙版本：firewall-cmd --version
更多防火墙状态：systemctl status firewalld.service
开启防火墙：systemctl start firewalld.service
关闭防火墙：systemctl stop firewalld.service
重启防火墙：systemctl restart firewalld.service
设置开机自启：systemctl enable firewalld.service
设置开机禁启：systemctl disable firewalld.service
检察是否自启：systemctl is-enabled firewalld.service
检察自启列表：systemctl list-unit-files|grep enabled
————————————————
五、Local运行模式环境

1、设置放开权限

    进入flink安装目录下的conf目录下修改flink-conf.yaml文件
    修改rest.bind-address： localhost 为 rest.bind-address：0.0.0.0，这个非常关键，是为了能在任何地方，比如windows的浏览器上访问flink web ui。否则访问失败，如下图：

2、启动flink本地模式

    local运行模式，主要用于测试，进入flink安装目录下，实行以下命令：

1. bin/start-cluster.sh

复制代码

 3、访问web页面

     访问 http://192.168.56.128:8081 或
     访问 http://hadoop001:8081

六、Standallone Cluster 独立集群 

1、设置放开权限

    进入flink安装目录下的conf目录下修改flink-conf.yaml文件
    修改rest.bind-address： localhost 为 rest.bind-address：0.0.0.0，这个非常关键，是为了能在任何地方，比如windows的浏览器上访问flink web ui。否则访问失败，如下图：

    修改jobmanager.bind-host: localhost 为jobmanager.bind-host：0.0.0.0，放开flink的rpc通讯权限，等心跳包发送到jobmanager则注册乐成，否则Total Task Slots  即是0     Task Managers 即是0   此时若提交任务,直接报资源不可用，如下图：

2、指定主节点ip地址

1. vi /usr/local/flink/conf/flink-conf.yaml
2. # (修改)指定主节点ip地址
3. jobmanager.rpc.address: hadoop001

复制代码

3、指定从节点

      指定从节点后，启动时候会自动在对应的从节点上启动TaskManager进程。

1. vi workers
2. # （修改）指定从节点
3. hadoop001
4. hadoop002
5. hadoop003

复制代码

4、指定主节点

     指定主节点，则启动时候，JobManager进程在指定的主节点上启动。假如不指定，则在哪台呆板上启动Flink，哪台呆板就是主节点，即JobManager进程就在哪台呆板上。    

1. vi masters
2. # 改成主节点hadoop001
3. hadoop001:8081

复制代码

 5、启动flink 的Standallone Cluster模式

     Standallone Cluster 独立集群（做实时盘算，不必要hadoop，该独立集群可能用的上）
     进入flink安装目 录下，实行以下命令：

1. bin/start-cluster.sh

复制代码

6、访问web页面

     访问 http://192.168.56.128:8081 或
     访问 http://hadoop001:8081

七、Flink提交任务快速启动

1、Idea创建项目flink

2、添加依赖

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.     <modelVersion>4.0.0</modelVersion>
7.     <groupId>org.example</groupId>
8.     <artifactId>flink</artifactId>
9.     <version>1.0-SNAPSHOT</version>
11.     <properties>
12.         <maven.compiler.source>8</maven.compiler.source>
13.         <maven.compiler.target>8</maven.compiler.target>
14.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
15.         <flink.version>1.17.0</flink.version>
16.     </properties>
18.     <dependencies>
19.         <dependency>
20.             <groupId>org.apache.flink</groupId>
21.             <artifactId>flink-streaming-java</artifactId>
22.             <version>${flink.version}</version>
23.         </dependency>
25.         <dependency>
26.             <groupId>org.apache.flink</groupId>
27.             <artifactId>flink-clients</artifactId>
28.             <version>${flink.version}</version>
29.         </dependency>
30.     </dependencies>
31.     <build>
32.         <plugins>
33.             <plugin>
34.                 <groupId>org.apache.maven.plugins</groupId>
35.                 <artifactId>maven-assembly-plugin</artifactId>
36.                 <version>3.0.0</version>
37.                 <configuration>
38.                     <descriptorRefs>
39.                         <descriptorRef>jar-with-dependencies</descriptorRef>
40.                     </descriptorRefs>
41.                 </configuration>
42.                 <executions>
43.                     <execution>
44.                         <id>make-assembly</id>
45.                         <phase>package</phase>
46.                         <goals>
47.                             <goal>single</goal>
48.                         </goals>
49.                     </execution>
50.                 </executions>
52.             </plugin>
53.         </plugins>
54.     </build>
55. </project>

复制代码

3、编写代码

1. package com.hwadee.flink;
4. import org.apache.flink.api.common.typeinfo.Types;
5. import org.apache.flink.api.java.tuple.Tuple2;
6. import org.apache.flink.api.java.utils.ParameterTool;
7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
8. import org.apache.flink.streaming.api.datastream.KeyedStream;
9. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
11. import org.apache.flink.util.Collector;
13. /**
14. * @ClassName$ FlinkBatchWordCount
15. * @Description 使用Flink编写一个批处理程序，实现单词统计功能
16. * <p>
17. * DataStream 批流统一处理，数据流的统一处理接口。
18. * 流处理中有两种不同的流：
19. * 批处理，是将其数据当作有界(有定义开始，有定义结束)流处理，例如文本文件数据
20. * 流处理，是将其数据当作无界(无定义开始，无定义结束)流处理，例如实时数据
21. *
22. * 实时数据： 打开集群，在某节点上进行 执行命令nc -lk 7777(指定未占用的端口号)，输入内容作为实时数据
23. *
24. * 首先设置 idea中的参数配置
25. * program params 中 设置 --host 192.168.56.128 --port 7777
26. */
27. public class FlinkRealTimeStreamWordCount03 {
28.     public static void main(String[] args) throws Exception {
29.         // 1、创建流处理执行环境
30.         StreamExecutionEnvironment env =
31.                 StreamExecutionEnvironment.getExecutionEnvironment();
32.         System.out.println("please wait for inpu data ...");
33.         // 从参数中提取主机名和端口号
34.         ParameterTool tool = ParameterTool.fromArgs(args);
35.         String hostname = tool.get("host");
36.         Integer port = tool.getInt("port");
37.         // 2、 读取数据，创建数据源
38.         DataStreamSource<String> source =
39.                 env.socketTextStream(hostname,port);
40.         // 3、 对数据进行转换处理
41.         SingleOutputStreamOperator<Tuple2<String, Long>> operator = source.flatMap(
42.                 (String line, Collector<Tuple2<String, Long>> out) -> {
43.                     // 将输入的文本进行分割
44.                     String[] words = line.split(" ");
45.                     // 将每个单词转换为 Tuple2输出
46.                     for (String word : words) {
47.                         if (word.contains(".")) {
48.                             out.collect(Tuple2.of(".", 1L));
49.                         }
50.                         out.collect(Tuple2.of(word, 1L));
51.                     }
52.                 }
53.         ).returns(Types.TUPLE(Types.STRING, Types.LONG));
54.         // 4、 数据按key值分组
55.         // 第一种方式已经弃用
56.         // KeyedStream<Tuple2<String, Long>, Tuple> keyBy = operator.keyBy(0);
57.         // 第二种方式推荐,使用 selector，同样使用 lambda表达式
58.         // operator.keyBy(data -> {return data.f0;}); 只有一行代码，大括号和return 可以省略
59.         KeyedStream<Tuple2<String, Long>, String> keyBy = operator.keyBy(data -> data.f0);
60.         // 5、 进行数据聚合
61.         SingleOutputStreamOperator<Tuple2<String, Long>> sum = keyBy.sum(1);
62.         // 6、 输出结果
63.         sum.print();
64.         // 7. 执行
65.         env.execute();
66.         /**
67.          * 结果说明
68.          * Idea 使用多线程模拟分布式Flink集群 运行并行任务
69.          * 前面数字代表线程号，在flink中是 slot ，即最小的单任务槽
70.          * 输出顺序是因为集群运行并行任务。
71.          * 注意 ： 只有在同一个任务上才能进行叠加。如 hello 叠加都是在 线程14上执行的。
72.          * 14> (hello,1)
73.          * 17> (It,1)
74.          * 10> (brain,,1)
75.          * 14> (hello,2)
76.          * 08> (Here,1)
77.          * 14> (hello,3)
78.          * 10> (Here,1)
79.          */
80.     }
81. }

复制代码

4、项目打包

5、添加任务及参数

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。