4.1.3 Flink-流处理框架-Flink-Local模式环境搭建

淩亂°似流年 2022-12-29 09:28 145阅读 0赞

**目录**

1.前置准备

2.安装Flink-Local模式环境

环境搭建

2.1 下载并解压

2.2 配置环境变量

2.3 shell启动测试

2.4 wordcount案例

3.提交任务

作业提交

3.1  启动Flink

3.2 提交作业

3.3 查看结果

--------------------

# 1.前置准备 #

CentOS7、jdk1.8、flink-1.10.1

# 2.安装Flink-Local模式环境 #

## 环境搭建 ##

### 2.1 下载并解压 ###

下载 `Flink` 安装包，这里我下载的是**`flink-1.10.1-bin-scala_2.11.tgz`**。下载地址：https://flink.apache.org/zh/downloads.html

# 解压
    [xiaokang@hadoop ~]$ tar -zxvf flink-1.10.1-bin-scala_2.11.tgz -C /opt/software/

### 2.2 配置环境变量 ###

[xiaokang@hadoop ~]$ sudo vim /etc/profile.d/env.sh

在原来基础上更新配置环境变量：

export FLINK_HOME=/opt/software/flink-1.10.1
    export PATH=${JAVA_HOME}/bin:${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:${ZOOKEEPER_HOME}/bin:${HIVE_HOME}/bin:${ZEPPELIN_HOME}/bin:${HBASE_HOME}/bin:${SQOOP_HOME}/bin:${FLUME_HOME}/bin:${PYTHON_HOME}/bin:${SCALA_HOME}/bin:${MAVEN_HOME}/bin:${GRADLE_HOME}/bin:${KAFKA_HOME}/bin:${SPARK_HOME}/bin:${FLINK_HOME}/bin:$PATH

执行 `source` 命令，使得配置的环境变量立即生效：

[xiaokang@hadoop ~]$ source /etc/profile.d/env.sh

### 2.3 shell启动测试 ###

[xiaokang@hadoop ~]$ start-scala-shell.sh local

![图片][935ae296f3fa8391764f4d38c004b745.png]

### 2.4 wordcount案例 ###

准备一个需要统计词频的小文件，部分词频数据：

xiaokang    Hive    Hadoop  Flink   微信公众号：小康新鲜事儿    MapReduce   Spark
    HBase    Spark   微信公众号：小康新鲜事儿
    HBase    Spark   微信公众号：小康新鲜事儿    Hadoop  MapReduce   Hive
    xiaokang    MapReduce   HBase   Flink   Spark   Hadoop
    Flink    Hadoop  Spark   xiaokang
    xiaokang    MapReduce   Hive    HBase   Spark
    微信公众号：小康新鲜事儿    HBase   MapReduce   Hive    Spark   xiaokang

Flink之批处理WordCount案例实操：

scala> val result=benv.readTextFile("file:///home/xiaokang/wordcount-xiaokang.txt").flatMap(_.split("\t")).map((_,1)).groupBy(0).sum(1)
    result1: org.apache.flink.api.scala.AggregateDataSet[(String, Int)] = org.apache.flink.api.scala.AggregateDataSet@4159c66d
    
    scala> result.print
    (Flink,617)
    (HBase,642)
    (Hadoop,644)
    (Hive,636)
    (MapReduce,631)
    (Spark,614)
    (xiaokang,647)
    (微信公众号：小康新鲜事儿,647)

查看 Flink Web UI 界面，端口为`8081`：

![图片][994ee415f5cba68e9a4b13de4ff16cf6.png]

# 3.提交任务 #

## 作业提交 ##

### 3.1  启动Flink ###

# 启动
    [xiaokang@hadoop ~]$ start-cluster.sh
    
    # 查看进程
    [xiaokang@hadoop ~]$ jps
    4533 Jps
    4456 TaskManagerRunner
    4122 StandaloneSessionClusterEntrypoint

### 3.2 提交作业 ###

使用安装包中自带的jar包来做流计算词频统计案例

# 1.虚拟机内开启端口
    nc -lk 1124
    # 2.提交作业
    [xiaokang@hadoop ~]$ flink run /opt/software/flink-1.10.1/examples/streaming/SocketWindowWordCount.jar --port 1124

该 jar 包的源码在 Flink 官方的 GitHub 仓库中可以找到，地址：https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/socket/SocketWindowWordCount.java ，可选传参有 hostname、port，对应的词频数据需要使用**空格**进行分割。

输入测试数据

[xiaokang@hadoop conf]$ nc -lk 1124
    xiaokang 微信公众号：小康新鲜事儿 xiaokangxxs xiaokang hive
    spark xiaokang hive hadoop flink
    flink xiaokang flink

### 3.3 查看结果 ###

通过 WEB UI 的控制台查看作业统运行情况：8081

![图片][fc95a06a3a2919d22a5a761304a02b2c.png]

通过 WEB 控制台查看到统计结果：

![图片][32ff26b30095f5c4dc34cc32a2327db1.png]

[935ae296f3fa8391764f4d38c004b745.png]: /images/20221120/ae126004d754409abfe88ca75a9f2127.png
[994ee415f5cba68e9a4b13de4ff16cf6.png]: /images/20221120/4ad5ce47812341ecaa5e48d19bd67a2e.png
[fc95a06a3a2919d22a5a761304a02b2c.png]: https://img-blog.csdnimg.cn/img_convert/fc95a06a3a2919d22a5a761304a02b2c.png
[32ff26b30095f5c4dc34cc32a2327db1.png]: https://img-blog.csdnimg.cn/img_convert/32ff26b30095f5c4dc34cc32a2327db1.png