Flink 环境的搭建、独立集群、Flink on Yarn、访问Flink web界面、Flink提 - ToB企服应用市场:ToB评测及商务社交产业平台

192.168.56.128 hadoop001
192.168.56.129 hadoop002
192.168.56.130 hadoop003

复制代码

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

复制代码

source /root/.bash_profile

复制代码

java -version

复制代码

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

复制代码

bin/start-cluster.sh

复制代码

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

复制代码

vi workers
# （修改）指定从节点
hadoop001
hadoop002
hadoop003

复制代码

vi masters
# 改成主节点hadoop001
hadoop001:8081

复制代码

bin/start-cluster.sh

复制代码

<?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</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flink.version>1.17.0</flink.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>${flink.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.0.0</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

复制代码

package com.hwadee.flink;
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.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;
/**
* @ClassName$ FlinkBatchWordCount
* @Description 使用Flink编写一个批处理程序，实现单词统计功能
* <p>
* DataStream 批流统一处理，数据流的统一处理接口。
* 流处理中有两种不同的流：
* 批处理，是将其数据当作有界(有定义开始，有定义结束)流处理，例如文本文件数据
* 流处理，是将其数据当作无界(无定义开始，无定义结束)流处理，例如实时数据
*
* 实时数据：打开集群，在某节点上进行执行命令nc -lk 7777(指定未占用的端口号)，输入内容作为实时数据
*
* 首先设置 idea中的参数配置
* program params 中设置 --host 192.168.56.128 --port 7777
*/
public class FlinkRealTimeStreamWordCount03 {
public static void main(String[] args) throws Exception {
// 1、创建流处理执行环境
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
System.out.println("please wait for inpu data ...");
// 从参数中提取主机名和端口号
ParameterTool tool = ParameterTool.fromArgs(args);
String hostname = tool.get("host");
Integer port = tool.getInt("port");
// 2、读取数据，创建数据源
DataStreamSource<String> source =
env.socketTextStream(hostname,port);
// 3、对数据进行转换处理
SingleOutputStreamOperator<Tuple2<String, Long>> operator = source.flatMap(
(String line, Collector<Tuple2<String, Long>> out) -> {
// 将输入的文本进行分割
String[] words = line.split(" ");
// 将每个单词转换为 Tuple2输出
for (String word : words) {
if (word.contains(".")) {
out.collect(Tuple2.of(".", 1L));
}
out.collect(Tuple2.of(word, 1L));
}
}
).returns(Types.TUPLE(Types.STRING, Types.LONG));
// 4、数据按key值分组
// 第一种方式已经弃用
// KeyedStream<Tuple2<String, Long>, Tuple> keyBy = operator.keyBy(0);
// 第二种方式推荐,使用 selector，同样使用 lambda表达式
// operator.keyBy(data -> {return data.f0;}); 只有一行代码，大括号和return 可以省略
KeyedStream<Tuple2<String, Long>, String> keyBy = operator.keyBy(data -> data.f0);
// 5、进行数据聚合
SingleOutputStreamOperator<Tuple2<String, Long>> sum = keyBy.sum(1);
// 6、输出结果
sum.print();
// 7. 执行
env.execute();
/**
* 结果说明
* Idea 使用多线程模拟分布式Flink集群运行并行任务
* 前面数字代表线程号，在flink中是 slot ，即最小的单任务槽
* 输出顺序是因为集群运行并行任务。
* 注意：只有在同一个任务上才能进行叠加。如 hello 叠加都是在线程14上执行的。
* 14> (hello,1)
* 17> (It,1)
* 10> (brain,,1)
* 14> (hello,2)
* 08> (Here,1)
* 14> (hello,3)
* 10> (Here,1)
*/
}
}

复制代码