Flink 叢集執行原理兼部署及Yarn執行模式深入剖析-Flink牛刀小試

YARN · 發表 2018-11-24 18:15:10

摘要：版權宣告：本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。版權宣告：禁止轉載，歡迎學習。QQ郵箱地址：[email protected]，如有任何問題，可隨時聯絡。 1 Fl...

版權宣告：本套技術專欄是作者（秦凱新）平時工作的總結和昇華，通過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和叢集環境容量規劃等內容，請持續關注本套部落格。版權宣告：禁止轉載，歡迎學習。QQ郵箱地址：[email protected]，如有任何問題，可隨時聯絡。

1 Flink的前世今生（生態很重要）

很多人可能都是在 2015 年才聽到 Flink 這個詞，其實早在 2008 年，Flink 的前身已經是柏林理工大學一個研究性專案，在 2014 被 Apache 孵化器所接受，然後迅速地成為了 ASF（Apache Software Foundation）的頂級專案之一。

Apache Flink is an open source platform for distributed stream and batch data
processing. Flink’s core is a streaming dataflow engine that provides data
distribution, communication, and fault tolerance for distributed computations
over data streams. Flink builds batch processing on top of the streaming engine,
overlaying native iteration support, managed memory, and program optimization.
複製程式碼

Apache Flink 是一個開源的分散式，高效能，高可用，準確的流處理框架。
主要由 Java 程式碼實現，提供Java 和scala介面。
支援實時流(stream)處理和批(batch)處理，批資料只是流資料的一個極限特例。
Flink原生支援了迭代計算、記憶體管理和程式優化。
Flink目前也在重力打造屬於自己的大資料生態。（FinkSQL , Flink ML ,Flink Gelly等）

2 吞吐量悖論

流處理和批處理的糾結選擇和不容水火，Flink通過靈活的執行引擎，能夠同時支援批處理任務與流處理任務，但是悖論是永遠存在的。

流處理：Flink以固定的快取塊為單位進行網路資料傳輸，使用者可以通過設定快取塊超時值指定快取塊的傳輸時機。如果快取塊的超時值為0，則Flink的資料傳輸方式類似上文所提到流處理系統的標準模型，此時系統可以獲得最低的處理延遲。
批處理：如果快取塊的超時值為無限大，則Flink的資料傳輸方式類似上文所提到批處理系統的標準模型，此時系統可以獲得最高的吞吐量。
靈活的祕密：快取塊的超時值也可以設定為0到無限大之間的任意值。快取塊的超時閾值越小，則Flink流處理執行引擎的資料處理延遲越低，但吞吐量也會降低，反之亦然。通過調整快取塊的超時閾值，使用者可根據需求靈活地權衡系統延遲和吞吐量。

3 容錯的抉擇（Flink or Spark）

SparkStreaming ：微批次模型，EOS語義，基於RDD Checkpoint進行容錯，基於checkpoint狀態管理。狀態的狀態操作基於DStream模板進行管理，延時中等水平，吞吐量很高。詳情請參考我的SparkStreaming原始碼解讀。

Flink ：流處理模型，EOS語義，基於兩種狀態管理進行容錯，即：State和checkpoint兩種機制。狀態操作可以細粒化到運算元等操作上。延時不僅低，而且吞吐量也非常高。

- State基於task和operator兩種狀態。State型別進一步細分為
Keyed State和 Operator State 兩種型別
- checkpoint基全域性快照來實現資料容錯，注意：State的狀態儲存在java的堆裡面，
checkpoint則通過定時實現全域性（所有State）狀態的持久化。
複製程式碼

說實在的，Flink很狂妄：

4 Stanalone 環境全方位剖析

4.1 Stanalone 模式

叢集節點規劃(一主兩從)

1 基礎環境：

jdk1.8及以上【需要配置JAVA_HOME】
ssh免密碼登入(至少要實現主節點能夠免密登入到從節點)
主機名hostname
/etc/hosts檔案配置主機名和ip的對映關係
192.168.1.160SparkMaster
192.168.1.161SparkWorker1
192.168.1.162SparkWorker2
關閉防火牆
複製程式碼

2 在SparkMaster節點上主要需要修改的配置資訊

cd /usr/local/flink-1.6.1/conf
vi flink-conf.yaml
jobmanager.rpc.address: SparkMaster
複製程式碼

3 slaves修改

vi slaves
SparkWorker1
SparkWorker2
複製程式碼

4 然後再把修改好的flink目錄拷貝到其他兩個節點即可

scp -rq flink-1.6.1 SparkWorker1:/usr/local/
scp -rq flink-1.6.1 SparkWorker2:/usr/local/
複製程式碼

4.2 Stanalone 執行展示

這裡發生一個小插曲，因為yarn配置檔案不一致，導致 hadoop Web UI 無法正常顯示所有NodeManager。所以注意配置檔案的一致性。

SparkMaster節點程序：

14273 SecondaryNameNode
15010 Worker
14038 DataNode
25031 StandaloneSessionClusterEntrypoint
13895 NameNode
14903 Master
14424 ResourceManager
14569 NodeManager
25130 Jps
複製程式碼

SparkWorker節點程序：

5732 Worker
10420 NodeManager
10268 DataNode
10540 Jps
8351 TaskManagerRunner
複製程式碼

上圖一張：

4.3 Stanalone 簡單任務測試

(1) 增量聚合: 視窗中每進入一條資料，就進行一次計算

實現方法主要有：

reduce(reduceFunction) aggregate(aggregateFunction) sum(),min(),max()

(2) 全量聚合: 等於視窗內的資料到齊，才開始進行聚合計算

全量聚合：可以實現對視窗內的資料進行排序等需

實現方法主要有：

apply(windowFunction)
 process(processWindowFunction)
 processWindowFunction比windowFunction提供了更多的上下文資訊。
複製程式碼

全量聚合詳細案例如下：

public class SocketDemoFullCount {

public static void main(String[] args) throws Exception{
//獲取需要的埠號
int port;
try {
ParameterTool parameterTool = ParameterTool.fromArgs(args);
port = parameterTool.getInt("port");
}catch (Exception e){
System.err.println("No port set. use default port 9010--java");
port = 9010;
}

//獲取flink的執行環境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

String hostname = "SparkMaster";
String delimiter = "\n";

//連線socket獲取輸入的資料
DataStreamSource<String> text = env.socketTextStream(hostname, port, delimiter);

DataStream<Tuple2<Integer,Integer>> intData = text.map(new MapFunction<String, Tuple2<Integer,Integer>>() {
@Override
public Tuple2<Integer,Integer> map(String value) throws Exception {
return new Tuple2<>(1,Integer.parseInt(value));
}
});

intData.keyBy(0)
.timeWindow(Time.seconds(5))
.process(new ProcessWindowFunction<Tuple2<Integer,Integer>, String, Tuple, TimeWindow>() {
@Override
public void process(Tuple key, Context context, Iterable<Tuple2<Integer, Integer>> elements, Collector<String> out)
throws Exception {
System.out.println("執行process......");
long count = 0;
for(Tuple2<Integer,Integer> element: elements){
count++;
}
out.collect("window:"+context.window()+",count:"+count);
}
}).print();
//這一行程式碼一定要實現，否則程式不執行
env.execute("Socket window count");
}
複製程式碼

}

(3) 資料來源

root@SparkMaster:/usr/local/install/hadoop-2.7.3/sbin# nc -l 9010
12
1
複製程式碼

(4) 執行結果

4.4 Stanalone 引數調優設定

引數調優設定:
1.jobmanager.heap.mb：jobmanager節點可用的記憶體大小
2.taskmanager.heap.mb：taskmanager節點可用的記憶體大小
3.taskmanager.numberOfTaskSlots：每臺機器可用的cpu數量
4.parallelism.default：預設情況下任務的並行度
5.taskmanager.tmp.dirs：taskmanager的臨時資料儲存目錄

slot和parallelism總結:
1.slot是靜態的概念，是指taskmanager具有的併發執行能力
2.parallelism是動態的概念，是指程式執行時實際使用的併發能力
3.設定合適的parallelism能提高運算效率，太多了和太少了都不行
複製程式碼

4.5 Stanalone 叢集啟動與掛機

啟動jobmanager
如果叢集中的jobmanager程序掛了，執行下面命令啟動。
bin/jobmanager.sh start
bin/jobmanager.sh stop
啟動taskmanager
新增新的taskmanager節點或者重啟taskmanager節點
bin/taskmanager.sh start
bin/taskmanager.sh stop
複製程式碼

5 資源排程環境（Yarn 模式）

5.1 模式1：（常駐session）

開闢資源 yarn - session . sh

1啟動一個一直執行的flink叢集
./bin/yarn-session.sh -n 2 -jm 1024 -tm 1024 -d

2 附著到一個已存在的flink yarn session
./bin/yarn-session.sh -id application_1463870264508_0029

3 資源所在地/tmp/.yarn-properties-root.
2018-11-24 17:24:19,644 INFOorg.apache.flink.yarn.cli.FlinkYarnSessionCli
- Found Yarn properties file under /tmp/.yarn-properties-root.

4：yarn資源描述
root@SparkMaster:/usr/local/install/flink-1.6.1# vim /tmp/.yarn-properties-root

#Generated YARN properties file
#Sat Nov 24 17:39:07 CST 2018
parallelism=2
dynamicPropertiesString=
applicationID=application_1543052238521_0001
複製程式碼

執行任務flink run

3 執行任務
hadoop fs -mkdir /input/
hadoop fs -put README.txt/input/

./bin/flink run ./examples/batch/WordCount.jar -input hdfs://SparkMaster:9000/input/README.txt -output hdfs://SparkMaster:9000/wordcount-result.txt 

4：執行結果
root@SparkMaster:/usr/local/install/flink-1.6.1# hadoop fs -cat/wordcount-result.txt

1 1
13 1
5d002 1
740 1
about 1
account 1
administration 1
複製程式碼

停止任務 【web介面或者命令列執行cancel命令】
複製程式碼

5.2 模式2：（session獨立互不影響）

1 啟動叢集，執行任務

./bin/flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar-input hdfs://SparkMaster:9000/input/README.txt -output hdfs://SparkMaster:9000/wordcount-result6.txt 

2018-11-24 17:56:18,066 INFOorg.apache.flink.yarn.AbstractYarnClusterDescriptor- Waiting for the cluster to be allocated
2018-11-24 17:56:18,078 INFOorg.apache.flink.yarn.AbstractYarnClusterDescriptor- Deploying cluster, current state ACCEPTED
2018-11-24 17:56:24,901 INFOorg.apache.flink.yarn.AbstractYarnClusterDescriptor- YARN application has been deployed successfully.
複製程式碼

2 ：提交一次，生成一個Yarn-session

6 flink run 引數指定:

1 引數必選 ： 
-n,--container <arg>分配多少個yarn容器 (=taskmanager的數量)
2 引數可選 ： 
-D <arg>動態屬性
-d,--detached獨立執行
-jm,--jobManagerMemory <arg>JobManager的記憶體 [in MB]
-nm,--name在YARN上為一個自定義的應用設定一個名字
-q,--query顯示yarn中可用的資源 (記憶體, cpu核數)
-qu,--queue <arg>指定YARN佇列.
-s,--slots <arg>每個TaskManager使用的slots數量
-tm,--taskManagerMemory <arg>每個TaskManager的記憶體 [in MB]
-z,--zookeeperNamespace <arg>針對HA模式在zookeeper上建立NameSpace 
-id,--applicationId <yarnAppId> YARN叢集上的任務id，附著到一個後臺執行的yarn session中

3 run [OPTIONS] <jar-file> <arguments>

run操作引數:
-c,--class <classname>如果沒有在jar包中指定入口類，則需要在這裡通過這個引數指定
-m,--jobmanager <host:port>指定需要連線的jobmanager(主節點)地址，使用這個引數可以指定一個不同於配置檔案中的jobmanager
-p,--parallelism <parallelism>指定程式的並行度。可以覆蓋配置檔案中的預設值。

4 啟動一個新的yarn-session,它們都有一個y或者yarn的字首

例如：./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar 

連線指定host和port的jobmanager：
./bin/flink run -m SparkMaster:1234 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1

啟動一個新的yarn-session：
./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1

5 注意：命令列的選項也可以使用./bin/flink 工具獲得。

6 Action "run" compiles and runs a program.

Syntax: run [OPTIONS] <jar-file> <arguments>
"run" action options:
-c,--class <classname>Class with the program entry point
("main" method or "getPlan()" method.
Only needed if the JAR file does not
specify the class in its manifest.
-C,--classpath <url>Adds a URL to each user code
classloaderon all nodes in the
cluster. The paths must specify a
protocol (e.g. file://) and be
accessible on all nodes (e.g. by means
of a NFS share). You can use this
option multiple times for specifying
more than one URL. The protocol must
be supported by the {@link
java.net.URLClassLoader}.
-d,--detachedIf present, runs the job in detached
mode
-n,--allowNonRestoredStateAllow to skip savepoint state that
cannot be restored. You need to allow
this if you removed an operator from
your program that was part of the
program when the savepoint was
triggered.
-p,--parallelism <parallelism>The parallelism with which to run the
program. Optional flag to override the
default value specified in the
configuration.
-q,--sysoutLoggingIf present, suppress logging output to
standard out.
-s,--fromSavepoint <savepointPath>Path to a savepoint to restore the job
from (for example
hdfs:///flink/savepoint-1537).

7Options for yarn-cluster mode:
-d,--detachedIf present, runs the job in detached
mode
-m,--jobmanager <arg>Address of the JobManager (master) to
which to connect. Use this flag to
connect to a different JobManager than
the one specified in the
configuration.
-yD <property=value>use value for given property
-yd,--yarndetachedIf present, runs the job in detached
mode (deprecated; use non-YARN
specific option instead)
-yh,--yarnhelpHelp for the Yarn session CLI.
-yid,--yarnapplicationId <arg>Attach to running YARN session
-yj,--yarnjar <arg>Path to Flink jar file
-yjm,--yarnjobManagerMemory <arg>Memory for JobManager Container with
optional unit (default: MB)
-yn,--yarncontainer <arg>Number of YARN container to allocate
(=Number of Task Managers)
-ynl,--yarnnodeLabel <arg>Specify YARN node label for the YARN
application
-ynm,--yarnname <arg>Set a custom name for the application
on YARN
-yq,--yarnqueryDisplay available YARN resources
(memory, cores)
-yqu,--yarnqueue <arg>Specify YARN queue.
-ys,--yarnslots <arg>Number of slots per TaskManager
-yst,--yarnstreamingStart Flink in streaming mode
-yt,--yarnship <arg>Ship files in the specified directory
(t for transfer)
-ytm,--yarntaskManagerMemory <arg>Memory per TaskManager Container with
optional unit (default: MB)
-yz,--yarnzookeeperNamespace <arg>Namespace to create the Zookeeper
sub-paths for high availability mode
-z,--zookeeperNamespace <arg>Namespace to create the Zookeeper
sub-paths for high availability mode
複製程式碼

6 結語

Flink 是一個是一個開源的分散式，高效能，高可用，準確的流處理框架。主要由 Java 程式碼實現。支援實時流(stream)處理和批(batch)處理，批資料只是流資料的一個極限特例。 Flink原生支援了迭代計算、記憶體管理和程式優化。本文立意在執行原理兼部署及Yarn執行模式，後續精彩內容請持續關注本部落格，辛苦成文，各自珍惜，謝謝！

秦凱新於深圳