使用Flink集群环境进行数据处理

前言

上篇文章记录了搭建分布式Flink集群环境的过程 搭建Flink集群环境

这篇文章咱们聊一聊Flink客户端如何对接Flink集群环境的过程

示例&＃xff1a;Flink读取Hadoop中的文件 然后通过集群环境进行数据处理的过程

Hadoop

Hadoop集群环境搭建

搭建大数据运行环境之一

搭建大数据运行环境之二

Hadoop集群端口说明

Hadoop集群搭建过程异常情况

不能格式化存储目录

详细异常信息

org.apache.hadoop.hdfs.qjournal.client.QuorumException: Could not format one or more JournalNodes. 1 exceptions thrown:
192.168.84.132:8485: Directory /usr/local/hadoop/jn/data/nameservices001 is in an inconsistent state: Can&＃39;t format the storage directory because the current directory is not empty


journalnode的端口是8485

处理方式

每一个hadoop journalnode节点上将指定目录删除即可

rm -rf /usr/local/hadoop/jn/data/nameservices001


上传文件到hdfs

cd /usr/local/hadoop/sbin
# 创建文件夹
hdfs dfs -mkdir /hdfsdata
# 文件
sudo vi /home/aaa.txt
# 上传文件到指定文件夹
hdfs dfs -put /home/aaa.txt  /hdfsdata


上传文件异常

Hadoop DataNode 节点启不来

详细异常信息

File /hdfsdata/aaa.txt._COPYING_ could only be replicated to 0 nodes instead of minReplication (&＃61;1). 
There are 0 datanode(s) running and no node(s) are excluded in this operation


查看WebUI DataNode情况

http://192.168.84.128:50070/dfshealth.html#tab-datanode


解决方法一

停止集群

cd /usr/local/hadoop/sbin
./stop-all.sh


删除在hdfs中配置的data目录

• 查看data目录

在core-site.xml中配置的hadoop.tmp.dir对应文件件

cat /usr/local/hadoop/etc/hadoop/core-site.xml 


• 删除

rm -rf /usr/local/hadoop/tmp/*


重新格式化

./hadoop namenode -format


重新启动集群

./start-all.sh


解决方法二

如果上面的方法还是不能启动DataNode那么使用这个方法

当执行文件系统格式化时
会在namenode数据文件夹
&＃xff08;即配置文件中dfs.name.dir在本地系统的路径&＃xff09;
中保存一个current/VERSION文件
记录namespaceID
标志了所有格式化的namenode版本
如果我们频繁的格式化namenode
那么datanode中保存&＃xff08;即dfs.data.dir在本地系统的路径&＃xff09;的current/VERSION文件只是你地第一次格式化时保存的namenode的ID
因此就会造成namenode和datanode之间的ID不一致


• 解决方法A&＃xff1a;&＃xff08;推荐&＃xff09;

删除DataNode的所有资料及将集群中每个datanode节点的/dfs/data/current中的VERSION删除
然后重新执行hadoop namenode -format进行格式化
重启集群&＃xff0c;错误消失


• 解决方法B&＃xff1a;

将name/current下的VERSION中的clusterID复制到data/current下的VERSION中&＃xff0c;覆盖掉原来的clusterID


查看DataNode情况

DataNode已经起来了

查看上传文件

http://192.168.84.128:50070


该文件路径

hdfs://192.168.84.128:8020/hdfsdata/aaa.txt


Flink读取数据源并处理数据

DEMO源码

https://gitee.com/pingfanrenbiji/flink-examples-streaming


Flink读取hdfs文件并处理数据

创建flink执行环境

• 第一个参数&＃xff1a;远程flink集群 jobmanager ip地址
• 第二个参数&＃xff1a;8081是jobmanager webui端口
• 第三个参数&＃xff1a;是当前文件夹所在的jar包

数据源

读取hdfs文件数据


各种算子简介

以单词计数为例

先要将字符串数据解析成单词和次数 使用tuple2表示
第一个字段是单词 第二个字段是次数
次数初始值设置成1


flatmap

flatmap来做解析的工作
一行数据可能有多个单词



keyBy


将数据流按照单词字段即0号索引字段做分组
keyBy(int index) 得到一个以单词为key的tuple2数据流



timeWindow


在流上指定想要的窗口
并根据窗口中的数据计算结果
每5秒聚合一次单词数
每个窗口都是从零开始统计的

timeWindow 指定想要5秒的翻滚窗口(Tumble)



sum


第三个调用为每个key每个窗口指定了sum聚合函数
按照次数字段&＃xff08;即1号索引字段想家&＃xff09;
得到结果数据流
将每5秒输出一次 这5秒内每个单词出现的次数



将数据打印到控制台

所有算子操作&＃xff08;创建源、聚合、打印&＃xff09;只是构建了内部算子操作的图形

只有在execute被调用时才会在提交到集群或本地计算机上执行


执行报错 找不到代码异常

具体异常信息

Caused by: org.apache.flink.runtime.operators.util.CorruptConfigurationException: Could not read the user code wrapper: org.apache.flink.streaming.scala.examples.remotejob.RemoteJobTest$$anon$2


解决方法

• 将当前目录文件夹打包成jar包

使用maven插件maven-jar-plugin

• 第三个参数指向该jar包

在FLink Web UI查看该任务的执行过程

编译异常

无效的标记

--add-exports&＃61;java.base/sun.net.util&＃61;ALL-UNNAMED


不支持hdfs文件系统

具体异常信息

org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme &＃39;hdfs&＃39;. The scheme is not directly supported by Flink and no Hadoop file system to support this scheme could be loaded


处理方式

• 下载 flink hadoop资源jar包

https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.7.5-7.0/flink-shaded-hadoop-2-uber-2.7.5-7.0.jar


• 放入flink 安装包 lib目录下

每个节点都需要放上该jar包 然后重启flink集群环境

当前操作节点hadoop namenode节点为standby状态

具体详细信息

org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.ipc.StandbyException): Operation category READ is not supported in state standby


解决方法

重新格式化2个namenode节点即可

具体详见


搭建大数据运行环境之二

遗留问题

flink数据源来自于socket数据

启动socket服务并输入数据

问题是

Flink并没有监听到该socket数据
暂时还没有找到原因 
了解的朋友们请联系我 
指导我一下哦


如果本地环境是可以监听到的

后记

为了解决这个问题
我请教了下 “Apache Flink China社区”钉钉群里面的谢波老师
他告诉我&＃xff1a;

通过java或scala一般创建本地执行环境 即

&＃39;final StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();&＃39;

很少有

&＃39;final StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment(ip,port,jarfiles);&＃39;

这样用的

若使用flink分布式环境 那么通过web ui界面 上传jar包的方式来完成


这也就解释了为什么我没有找到相关资料
只能靠自己&＃39;摸着石头过河&＃39;了


结语

在了解一件新事物的时候 
按照自己的想法 一番努力和挣扎之后
也许方向是错误的 
但也会对它更进一步的了解了