Flink+hadoop部署及Demo

這篇具有很好參考價(jià)值的文章主要介紹了Flink+hadoop部署及Demo。希望對(duì)大家有所幫助。如果存在錯(cuò)誤或未考慮完全的地方，請(qǐng)大家不吝賜教，您也可以點(diǎn)擊"舉報(bào)違法"按鈕提交疑問(wèn)。

Hadoop集群高可用部署

下載hadoop包地址

https://dlcdn.apache.org/hadoop/common/hadoop-3.2.4/hadoop-3.2.4.tar.gz
上傳并解壓到3臺(tái)服務(wù)器
配置3臺(tái)主機(jī)的hosts和免密登錄
Flink+hadoop部署及Demo

1.修改.bash_profile

vi .bash_profile
# HADOOP_HOME
export HADOOP_HOME=/apps/svr/hadoop-3.2.4
export PATH=$PATH:$HADOOP_HOME/bin
export PATH=$PATH:$HADOOP_HOME/sbin

source .bash_profile
hadoop version查看hadoop下載版本
Flink+hadoop部署及Demo

架構(gòu)如下：

	hadoop01	hadoop02	hadoop03
HDFS	NameNode,DataNode	DataNode	NameNode,DataNode
YARN	ResouceManager,NodeManager	NodeManager	ResouceManager,NodeManager

NameNode 通過(guò)rpc高可用
ResouceManager通過(guò)zk高可用
主備目錄

在集群各節(jié)點(diǎn)創(chuàng)建目錄

mkdir -p /apps/svr/hadoop-3.2.4/tmp
mkdir -p /apps/svr/hadoop-3.2.4/dfs/name
mkdir -p /apps/svr/hadoop-3.2.4/dfs/data
mkdir -p /apps/svr/hadoop-3.2.4/journaldata

2.配置core-site.xml

        <property>
         <name>fs.defaultFS</name>
         <value>hdfs://mycluster/</value>
        </property>
        
        <!-- 指定hadoop工作目錄 -->
        <property>
          <name>hadoop.tmp.dir</name>
          <value>/apps/svr/hadoop-3.2.4/tmp</value>
        </property>
      
        <!-- 指定zookeeper集群訪問(wèn)地址 -->
        <property>
          <name>ha.zookeeper.quorum</name>
          <value>10.251.75.112:2181,10.251.75.113:2181,10.251.75.114:2181</value>
        </property>
      
        <!-- 配置為了解決以后其他組件連接HDFS集群  -->
        <property>
          <name>hadoop.proxyuser.bigdata.hosts</name>
          <value>*</value>
        </property>
      
        <property>
          <name>hadoop.proxyuser.bigdata.groups</name>
          <value>*</value>
        </property>

3.修改配置文件hdfs-site.xml

<!-- NameNode 存放的位置 -->
　　<property>
        <name>dfs.namenode.name.dir</name>
        <value>/apps/svr/hadoop-3.2.4/dfs/name</value>
    </property>
　　<!-- DataNode 存放的位置 -->
    <property>
        <name>dfs.datanode.data.dir</name>
        <value>/apps/svr/hadoop-3.2.4/dfs/data</value>
    </property>
<!-- 指定HDFS的nameservices為mycluster,需要跟core-site.xml中保持一致 -->
  <property>

    <name>dfs.replication</name>

    <value>2</value>

  </property>
<!-- 指定HDFS的nameservices為mycluster,需要跟core-site.xml中保持一致 -->
  <property>
          <name>dfs.nameservices</name>
          <value>mycluster</value>
  </property>
      
        <!-- 設(shè)置mycluster集群有兩個(gè)namenode, 分別為nn1,nn2 -->
  <property>
          <name>dfs.ha.namenodes.mycluster</name>
          <value>nn1,nn2</value>
  </property>
      
        <!-- 配置nn1 的RPC通信地址 -->
   <property>
          <name>dfs.namenode.rpc-address.mycluster.nn1</name>
          <value>10.251.75.112:9000</value>
   </property>
        
        <!-- 配置nn1的http通信地址 -->
   <property>
          <name>dfs.namenode.http-address.mycluster.nn1</name>
          <value>10.251.75.112:50070</value>
   </property>
 
 <!-- 配置nn2 的RPC通信地址 -->
   <property>
          <name>dfs.namenode.rpc-address.mycluster.nn2</name>
          <value>10.251.75.114:9000</value>
   </property>
      
        <!-- 配置nn2的http通信地址 -->
   <property>
          <name>dfs.namenode.http-address.mycluster.nn2</name>
          <value>10.251.75.114:50070</value>
   </property>
      
        <!-- 指定JournalNode 在本地磁盤存放數(shù)據(jù)的位置 -->
   <property>
          <name>dfs.journalnode.edits.dir</name>
          <value>/apps/svr/hadoop-3.2.4/journaldata</value>
   </property>
      
        <!-- 指定NameNode的edits元數(shù)據(jù)在journalNode上的服務(wù)器 -->
   <property>
          <name>dfs.namenode.shared.edits.dir</name>
          <value>qjournal://10.251.75.112:8485;10.251.75.113:8485;10.251.75.114:8485/mycluster</value>
   </property>
 
 <!-- 開(kāi)啟NameNode 自動(dòng)切換 -->
   <property>
          <name>dfs.ha.automatic-failover.enabled</name>
          <value>true</value>
   </property>
      
        <!-- 配置nameNode失敗自動(dòng)切換的實(shí)現(xiàn)方式 -->
   <property>
          <name>dfs.client.failover.proxy.provider.mycluster</name>
          <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>
   </property>
      
        <!-- 配置隔離機(jī)制方法 -->
   <property>
          <name>dfs.ha.fencing.methods</name>
          <value>
            sshfence
            shell(/bin/true)
          </value>
   </property>
      
        <!-- 使用sshfence隔離機(jī)制時(shí)需要ssh免密登陸 -->
   <property>
          <name>dfs.ha.fencing.ssh.private-key-files</name>
          <value>/apps/.ssh/id_rsa</value>
   </property>
 
  <!-- 配置sshfence隔離機(jī)制超時(shí)時(shí)間 -->
   <property>
          <name>dfs.ha.fencing.ssh.connect-timeout</name>
          <value>30000</value>
   </property>
     
   <property>
         <name>dfs.webhdfs.enabled</name>
         <value>true</value>
   </property>

4.配置mapred-site.xml

    <!-- 指定mapreduce運(yùn)算時(shí)資源調(diào)度為 yarn 模式 -->
       <property>
          <name>mapreduce.framework.name</name>
         <value>yarn</value>
        </property>
      
        <!-- 配置mapreduce歷史服務(wù)器地址 端口號(hào) -->
        <property>
          <name>mapreduce.jobhistory.address</name>
          <value>10.251.75.112:10020</value>
        </property>
      
        <!-- 配置mapreduce歷史服務(wù)器WEB訪問(wèn)地址 -->
        <property>
          <name>mapreduce.jobhistory.webapp.address</name>
          <value>10.251.75.112:19888</value>
        </property>

5.配置yarn-site.xml

<!-- Site specific YARN configuration properties -->
 
 <!-- 開(kāi)啟高可用 -->
  <property>
    <name>yarn.resourcemanager.ha.enabled</name>
    <value>true</value>
  </property>
 
  <!-- 指定ResourceManager的標(biāo)識(shí)：yrc -->
  <property>
    <name>yarn.resourcemanager.cluster-id</name>
    <value>yrc</value>
  </property>
 
  <!-- 指定RM的名字-->
  <property>
    <name>yarn.resourcemanager.ha.rm-ids</name>
    <value>rm1,rm2</value>
  </property>
 
 <!-- 指定rm1服務(wù)器 -->
  <property>
    <name>yarn.resourcemanager.hostname.rm1</name>
    <value>10.251.75.112</value>
  </property>
 
  <!-- 指定rm2服務(wù)器 -->
  <property>
    <name>yarn.resourcemanager.hostname.rm2</name>
    <value>10.251.75.114</value>
  </property>
 
  <!-- 指定rm 被管理的zk 地址 -->
  <property>
    <name>yarn.resourcemanager.zk-address</name>
    <value>10.251.75.112:2181,10.251.75.113:2181,10.251.75.114:2181</value>
  </property>
 
<!-- 運(yùn)行mapreduce任務(wù)需要使用的服務(wù) -->
  <property>
    <name>yarn.nodemanager.aux-services</name>
    <value>mapreduce_shuffle</value>
  </property>
 
  <!-- 開(kāi)啟yarn集群的日志聚合功能 -->
  <property>
    <name>yarn.log-aggregation-enable</name>
    <value>true</value>
  </property>
 
  <!-- 設(shè)置日志保存時(shí)間 -->
  <property>
    <name>yarn.log-aggregation.retain-seconds</name>
    <value>86400</value>
  </property>
 
  <!-- 啟動(dòng)rm自動(dòng)恢復(fù)功能 -->
  <property>
    <name>yarn.resourcemanager.recovery.enabled</name>
    <value>true</value>
  </property>
 
 <!-- 制定rm 狀態(tài)信息存儲(chǔ)在zookeeper集群上 -->
  <property>
    <name>yarn.resourcemanager.store.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value>
  </property>
  <!--是否啟動(dòng)一個(gè)線程檢查每個(gè)任務(wù)正使用的物理內(nèi)存量，如果任務(wù)超出分配值，則直接將其殺掉，默認(rèn)是true -->
  <!--是否啟動(dòng)一個(gè)線程檢查每個(gè)任務(wù)正使用的物理內(nèi)存量，如果任務(wù)超出分配值，則直接將其殺掉，默認(rèn)是true -->
  <property>
          <name>yarn.nodemanager.pmem-check-enabled</name>
          <value>false</value>
  </property>
 <!--是否啟動(dòng)一個(gè)線程檢查每個(gè)任務(wù)正使用的虛擬內(nèi)存量，如果任務(wù)超出分配值，則直接將其殺掉，默認(rèn)是true -->
  <property>
          <name>yarn.nodemanager.vmem-check-enabled</name>
          <value>false</value>
  </property>

6.修改文件workers文件

10.251.75.112
10.251.75.113
10.251.75.114

7.修改hadoop-env.sh

配置JAVA_HOME和主機(jī)上的jdk環(huán)境變量一樣

6.將core-site.xml,hdfs-site.xml,mapred-site.xml,yarn-site.xml傳到另外2臺(tái)主機(jī)

scp core-site.xml apps@10.251.75.113:/apps/svr/hadoop-3.2.4/etc/hadoop
scp hdfs-site.xml apps@10.251.75.113:/apps/svr/hadoop-3.2.4/etc/hadoop
scp mapred-site.xml apps@10.251.75.113:/apps/svr/hadoop-3.2.4/etc/hadoop
scp yarn-site.xml apps@10.251.75.113:/apps/svr/hadoop-3.2.4/etc/hadoop
scp workers apps@10.251.75.113:/apps/svr/hadoop-3.2.4/etc/hadoop
scp hadoop-env.sh apps@10.251.75.113:/apps/svr/hadoop-3.2.4/etc/hadoop

scp core-site.xml apps@10.251.75.114:/apps/svr/hadoop-3.2.4/etc/hadoop
scp hdfs-site.xml apps@10.251.75.114:/apps/svr/hadoop-3.2.4/etc/hadoop
scp mapred-site.xml apps@10.251.75.114:/apps/svr/hadoop-3.2.4/etc/hadoop
scp yarn-site.xml apps@10.251.75.114:/apps/svr/hadoop-3.2.4/etc/hadoop
scp workers apps@10.251.75.114:/apps/svr/hadoop-3.2.4/etc/hadoop
scp hadoop-env.sh apps@10.251.75.114:/apps/svr/hadoop-3.2.4/etc/hadoop

啟動(dòng)集群各個(gè)節(jié)點(diǎn)監(jiān)控NameNode的管理日志的JournalNode
在各節(jié)點(diǎn)執(zhí)行
hdfs --daemon start journalnode
Flink+hadoop部署及Demo
在node01上格式化namenode
hdfs namenode -format
在node1上啟動(dòng)namenode
hdfs --daemon start namenode

同步nn2 及node03上同步nn1及node01的源數(shù)據(jù)
hdfs namenode -bootstrapStandby
在node1節(jié)點(diǎn)上格式化ZKFC
hdfs zkfc -formatZK

node1節(jié)點(diǎn)上啟動(dòng)HDFS和Yarn
start-dfs.sh
Flink+hadoop部署及Demo
查看zk，namenode在node03上

查看hdfs管理頁(yè)面

http://10.251.75.112:50070/
http://10.251.75.114:50070/
Flink+hadoop部署及Demo

start-yarn.sh
進(jìn)程在node01和node03上
Flink+hadoop部署及Demo

查看HA zk節(jié)點(diǎn)數(shù)據(jù) resourceMaster在rm2上
Flink+hadoop部署及Demo
啟動(dòng)后查看yarn頁(yè)面 node03頁(yè)面
http://10.251.75.114:8088/cluster

Flink部署

下載包地址

flink下載地址
https://www.apache.org/dyn/closer.lua/flink/flink-1.16.1/flink-1.16.1-bin-scala_2.12.tgz

節(jié)點(diǎn)服務(wù)器	hadoop01	hadoop02	hadoop03
角色	JobManager

解壓后
配置source

export FLINK_HOME=/apps/svr/flink-1.16.1
export PATH=export PATH=$FLINK_HOME/bin:$PATH

source .bash_profile

Flink Per-Job模式

Per-Job 模式是指每個(gè)Flink Job都是一組獨(dú)立集群，即有自己的JobManager和TaskManager。提交任務(wù)后，YARN首先會(huì)為該任務(wù)分派一個(gè)AM容器，該容器內(nèi)會(huì)運(yùn)行一個(gè)JobManager進(jìn)程，之后JobManager會(huì)向Yarn申請(qǐng)運(yùn)行TaskManager所需要的container，container的數(shù)量和container的配置（CPU、內(nèi)存）會(huì)基于客戶端的需求來(lái)確定，當(dāng)JobManager和TaskManager進(jìn)程都拉起來(lái)之后，則會(huì)執(zhí)行相應(yīng)的Flink Job。這樣，與Standalone以及yarn-session不同的是，我們不需要準(zhǔn)備一個(gè)常駐的Flink 集群進(jìn)程，只需要保證本地有一個(gè)Flink環(huán)境即可，F(xiàn)link集群是在我們提交Job時(shí)動(dòng)態(tài)創(chuàng)建出來(lái)的。
這種方式的優(yōu)勢(shì)在于每個(gè)任務(wù)獨(dú)立運(yùn)行，相互不會(huì)收到干擾，這個(gè)不干擾還包括了運(yùn)行日志也是隔離的。另外，借助YARN集群的能力，提供了對(duì)Flink Job的全方位保障，包括JobManager的高可用，TaskManager的恢復(fù)，然后再結(jié)合Flink自身提供的健壯性，包括檢查點(diǎn)、保存點(diǎn)機(jī)制，從而能夠很好的保障Flink Job的高可用和健壯性。劣勢(shì)的話，就是保證了資源隔離的同時(shí)也占用了更多的資源，因?yàn)槊總€(gè)Job都需要一個(gè)JobManager，每個(gè)JobManager都會(huì)消耗一個(gè)AM進(jìn)程資源。

-yn,--container <arg> 表示分配容器的數(shù)量，也就是 TaskManager 的數(shù)量。
-d,--detached：設(shè)置在后臺(tái)運(yùn)行。
-yjm,--jobManagerMemory<arg>:設(shè)置 JobManager 的內(nèi)存，單位是 MB。
-ytm，--taskManagerMemory<arg>:設(shè)置每個(gè) TaskManager 的內(nèi)存，單位是 MB。
-ynm,--name:給當(dāng)前 Flink application 在 Yarn 上指定名稱。
-yq,--query：顯示 yarn 中可用的資源（內(nèi)存、cpu 核數(shù)）
-yqu,--queue<arg> :指定 yarn 資源隊(duì)列
-ys,--slots<arg> :每個(gè) TaskManager 使用的 Slot 數(shù)量。
-yz,--zookeeperNamespace<arg>:針對(duì) HA 模式在 Zookeeper 上創(chuàng)建 NameSpace
-yid,--applicationID<yarnAppId> : 指定 Yarn 集群上的任務(wù) ID,附著到一個(gè)后臺(tái)獨(dú) 立運(yùn)行的 Yarn Session 中。

使用flink run -m yarn-cluster --help 可查看可用命令

Flink+hadoop部署及Demo

Demo01

在flink上執(zhí)行命令跑demo
flink run -m yarn-cluster -t yarn-per-job -yjm 1024 -ytm 1024 /apps/svr/flink-1.16.1/examples/streaming/WordCount.jar
任務(wù)執(zhí)行完成并且
Flink+hadoop部署及Demo
通過(guò)yarn頁(yè)面查看到任務(wù)已完成，并且hdfs上有記錄

Demo02

登錄10.251.75.112 nc -lk 9999開(kāi)啟服務(wù)端端口監(jiān)聽(tīng)
nc -kl 9999
Flink+hadoop部署及Demo
flink run -m yarn-cluster -t yarn-per-job -yjm 1024 -ytm 1024 /apps/svr/flink-1.16.1/examples/streaming/SocketWindowWordCount.jar --hostname 10.251.75.112 --port 9999

打開(kāi)flink集群管理頁(yè)面

在服務(wù)端nc窗口輸入hello word即可看到flink管理頁(yè)面有對(duì)應(yīng)輸出
Flink+hadoop部署及Demo

在flink taskmanager即可看到對(duì)應(yīng)輸出
Flink+hadoop部署及Demo

將任務(wù)關(guān)閉

查看管理頁(yè)面
Flink+hadoop部署及Demo

yarn application -kill application_1684908112422_0008

查看管理頁(yè)面任務(wù)已被關(guān)閉
文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-466921.html