Flink執(zhí)行流程與源碼分析
本文轉(zhuǎn)載自微信公眾號(hào)「大數(shù)據(jù)左右手」,作者王了個(gè)博。轉(zhuǎn)載本文請(qǐng)聯(lián)系大數(shù)據(jù)左右手公眾號(hào)。
Flink主要組件
作業(yè)管理器(JobManager)
(1) 控制一個(gè)應(yīng)用程序執(zhí)行的主進(jìn)程,也就是說(shuō),每個(gè)應(yīng)用程序 都會(huì)被一個(gè)不同的Jobmanager所控制執(zhí)行
(2) Jobmanager會(huì)先接收到要執(zhí)行的應(yīng)用程序,這個(gè)應(yīng)用程序會(huì)包括:作業(yè)圖( Job Graph)、邏輯數(shù)據(jù)流圖( ogical dataflow graph)和打包了所有的類(lèi)、庫(kù)和其它資源的JAR包。
(3) Jobmanager會(huì)把 Jobgraph轉(zhuǎn)換成一個(gè)物理層面的 數(shù)據(jù)流圖,這個(gè)圖被叫做 “執(zhí)行圖”(Executiongraph),包含了所有可以并發(fā)執(zhí)行的任務(wù)。Job Manager會(huì)向資源管理器( Resourcemanager)請(qǐng)求執(zhí)行任務(wù)必要的資源,也就是 任務(wù)管理器(Taskmanager)上的插槽slot。一旦它獲取到了足夠的資源,就會(huì)將執(zhí)行圖分發(fā)到真正運(yùn)行它們的 Taskmanager上。而在運(yùn)行過(guò)程中Jobmanagera會(huì)負(fù)責(zé)所有需要中央?yún)f(xié)調(diào)的操作,比如說(shuō)檢查點(diǎn)(checkpoints)的協(xié)調(diào)。
任務(wù)管理器(Taskmanager)
(1) Flink中的工作進(jìn)程。通常在 Flink中會(huì)有多個(gè) Taskmanageria運(yùn)行, 每個(gè) Taskmanageri都包含了一定數(shù)量的插槽( slots)。插槽的數(shù)量限制了Taskmanageri能夠執(zhí)行的任務(wù)數(shù)量。
(2) 啟動(dòng)之后, Taskmanager會(huì)向資源管理器注冊(cè)它的插槽;收到資源管理器的指令后, Taskmanageri就會(huì)將一個(gè)或者多個(gè)插槽提供給Jobmanageri調(diào)用。Jobmanager就可以向插槽分配任務(wù)( tasks)來(lái)執(zhí)行了。
(3) 在執(zhí)行過(guò)程中, 一個(gè) Taskmanagera可以跟其它運(yùn)行同一應(yīng)用程序的Taskmanager交換數(shù)據(jù)。
資源管理器(Resource Manager)
(1) 主要負(fù)責(zé)管理任務(wù)管理器( Task Manager)的 插槽(slot)Taskmanger插槽是 Flink中定義的處理資源單元。
(2) Flink 為不同的環(huán)境和資源管理工具提供了不同資源管理器,比如YARNMesos、K8s,以及 standalone部署。
(3) 當(dāng) Jobmanager申請(qǐng)插槽資源時(shí), Resourcemanager會(huì)將有空閑插槽的Taskmanager?分配給Jobmanager。如果 Resourcemanagery沒(méi)有足夠的插槽來(lái)滿(mǎn)足 Jobmanager的請(qǐng)求, 它還可以向資源提供平臺(tái)發(fā)起會(huì)話,以提供啟動(dòng) Taskmanager進(jìn)程的容器。
分發(fā)器(Dispatcher)
(1) 可以跨作業(yè)運(yùn)行,它為應(yīng)用提交提供了REST接口。
(2)當(dāng)一個(gè)應(yīng)用被提交執(zhí)行時(shí),分發(fā)器就會(huì)啟動(dòng)并將應(yīng)用移交給Jobmanage
(3) Dispatcher他會(huì)啟動(dòng)一個(gè) WebUi,用來(lái)方便地 展示和監(jiān)控作業(yè)執(zhí)行的信息。
任務(wù)提交流程
- 提交應(yīng)用
- 啟動(dòng)并提交應(yīng)用
- 請(qǐng)求slots
- 任務(wù)啟動(dòng)
- 注冊(cè)slots
- 發(fā)出提供slot的指令
- 提供slots
- 提交要在slots中執(zhí)行的任務(wù)
- 交換數(shù)據(jù)
任務(wù)提交流程(YARN)
a. Flink任務(wù)提交后,Client向HDFS上傳Flink的Jar包和配置
b. 隨后向 Yarn ResourceManager提交任務(wù)ResourceManager分配 Container資源并通知對(duì)應(yīng)的NodeManager啟動(dòng)
c. ApplicationMaster,ApplicationMaster 啟動(dòng)后加載Flink的Jar包和配置構(gòu)建環(huán)境
d. 然后啟動(dòng)JobManager , 之后ApplicationMaster 向ResourceManager 申請(qǐng)資源啟動(dòng)TaskManager
e. ResourceManager 分配 Container 資源后 , 由ApplicationMaster通知資源所在節(jié)點(diǎn)的NodeManager啟動(dòng)TaskManager
f. NodeManager 加載 Flink 的 Jar 包和配置構(gòu)建環(huán)境并啟動(dòng) TaskManager
g. TaskManager 啟動(dòng)后向 JobManager 發(fā)送心跳包,并等待 JobManager 向其分配任務(wù)。
源碼分析--集群?jiǎn)?dòng) JobManager 啟動(dòng)分析
JobManager 的內(nèi)部包含非常重要的三大組件
- WebMonitorEndpoint
- ResourceManager
- Dispatcher
入口,啟動(dòng)主類(lèi):StandaloneSessionClusterEntrypoint
- // 入 口
- StandaloneSessionClusterEntrypoint.main() ClusterEntrypoint.runClusterEntrypoint(entrypoint);
- clusterEntrypoint.startCluster();
- runCluster(configuration, pluginManager);
- // 第一步:初始化各種服務(wù)
- /**
- * 初始化了 主節(jié)點(diǎn)對(duì)外提供服務(wù)的時(shí)候所需要的 三大核心組件啟動(dòng)時(shí)所需要的基礎(chǔ)服務(wù)
- * 初始化服務(wù),如 JobManager 的 Akka RPC 服務(wù),HA 服務(wù),心跳檢查服務(wù),metric service
- * 這些服務(wù)都是 Master 節(jié)點(diǎn)要使用到的一些服務(wù)
- * 1、commonRpcService: 基于 Akka 的 RpcService 實(shí)現(xiàn)。RPC 服務(wù)啟動(dòng) Akka 參與者來(lái)接收從 RpcGateway 調(diào)用 RPC
- * 2、haServices: 提供對(duì)高可用性所需的所有服務(wù)的訪問(wèn)注冊(cè),分布式計(jì)數(shù)器和領(lǐng)導(dǎo)人選舉
- * 3、blobServer: 負(fù)責(zé)偵聽(tīng)傳入的請(qǐng)求生成線程來(lái)處理這些請(qǐng)求。它還負(fù)責(zé)創(chuàng)建要存儲(chǔ)的目錄結(jié)構(gòu) blob 或臨時(shí)緩存它們
- * 4、heartbeatServices: 提供心跳所需的所有服務(wù)。這包括創(chuàng)建心跳接收器和心跳發(fā)送者。
- * 5、metricRegistry: 跟蹤所有已注冊(cè)的 Metric,它作為連接 MetricGroup 和 MetricReporter
- * 6、archivedExecutionGraphStore: 存儲(chǔ)執(zhí)行圖ExecutionGraph的可序列化形式。
- */
- initializeServices(configuration, pluginManager);
- // 創(chuàng)建 DispatcherResourceManagerComponentFactory, 初始化各種組件的
- 工廠實(shí)例
- // 其實(shí)內(nèi)部包含了三個(gè)重要的成員變量:
- // 創(chuàng)建 ResourceManager 的工廠實(shí)例
- // 創(chuàng)建 Dispatcher 的工廠實(shí)例
- // 創(chuàng)建 WebMonitorEndpoint 的工廠實(shí)例
- createDispatcherResourceManagerComponentFactory(configuration);
- // 創(chuàng)建 集群運(yùn)行需要的一些組件:Dispatcher, ResourceManager 等
- // 創(chuàng) 建 ResourceManager
- // 創(chuàng) 建 Dispatcher
- // 創(chuàng) 建 WebMonitorEndpoint
- clusterComponent = dispatcherResourceManagerComponentFactory.create(...)
1. initializeServices():初始化各種服務(wù)
- // 初 始 化 和 啟 動(dòng) AkkaRpcService, 內(nèi) 部 其 實(shí) 包 裝 了 一 個(gè) ActorSystem commonRpcService = AkkaRpcServiceUtils.createRemoteRpcService(...)
- // 初始化一個(gè)負(fù)責(zé) IO 的線程池
- ioExecutor = Executors.newFixedThreadPool(...)
- // 初始化 HA 服務(wù)組件,負(fù)責(zé) HA 服務(wù)的是:ZooKeeperHaServices haServices = createHaServices(configuration, ioExecutor);
- // 初始化 BlobServer 服務(wù)端
- blobServer = new BlobServer(configuration, haServices.createBlobStore()); blobServer.start();
- // 初始化心跳服務(wù)組件, heartbeatServices = HeartbeatServices heartbeatServices = createHeartbeatServices(configuration);
- // 初始化一個(gè)用來(lái)存儲(chǔ) ExecutionGraph 的 Store, 實(shí)現(xiàn)是:
- FileArchivedExecutionGraphStore
- archivedExecutionGraphStore = createSerializableExecutionGraphStore(...)
2. createDispatcherResourceManagerComponentFactory(configuration)初始化了多組件的工廠實(shí)例
- 1、DispatcherRunnerFactory,默認(rèn)實(shí)現(xiàn):DefaultDispatcherRunnerFactory
- 2、ResourceManagerFactory,默認(rèn)實(shí)現(xiàn):StandaloneResourceManagerFactory
- 3、RestEndpointFactory,默認(rèn)實(shí)現(xiàn):SessionRestEndpointFactory
- clusterComponent = dispatcherResourceManagerComponentFactory
- .create(configuration, ioExecutor, commonRpcService, haServices,
- blobServer, heartbeatServices, metricRegistry,
- archivedExecutionGraphStore,
- new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService()),
- this);
3. 創(chuàng)建 WebMonitorEndpoint
- /*************************************************
- * 創(chuàng)建 WebMonitorEndpoint 實(shí)例, 在 Standalone 模式下:DispatcherRestEndpoint
- * 1、restEndpointFactory = SessionRestEndpointFactory
- * 2、webMonitorEndpoint = DispatcherRestEndpoint
- * 3、highAvailabilityServices.getClusterRestEndpointLeaderElectionService() = ZooKeeperLeaderElectionService
- * 當(dāng)前這個(gè) DispatcherRestEndpoint 的作用是:
- * 1、初始化的過(guò)程中,會(huì)一大堆的 Handler
- * 2、啟動(dòng)一個(gè) Netty 的服務(wù)端,綁定了這些 Handler
- * 3、當(dāng) client 通過(guò) flink 命令執(zhí)行了某些操作(發(fā)起 restful 請(qǐng)求), 服務(wù)端由 webMonitorEndpoint 來(lái)執(zhí)行處理
- * 4、舉個(gè)例子: 如果通過(guò) flink run 提交一個(gè) Job,那么最后是由 webMonitorEndpoint 中的 JobSubmitHandler 來(lái)執(zhí)行處理
- * 5、補(bǔ)充一個(gè):job 由 JobSubmitHandler 執(zhí)行完畢之后,轉(zhuǎn)交給 Dispatcher 去調(diào)度執(zhí)行
- */
- webMonitorEndpoint = restEndpointFactory.createRestEndpoint(
- configuration, dispatcherGatewayRetriever, resourceManagerGatewayRetriever,
- blobServer, executor, metricFetcher,
- highAvailabilityServices.getClusterRestEndpointLeaderElectionService(),
- fatalErrorHandler
- );
4. 創(chuàng)建 resourceManager
- /*************************************************
- * 創(chuàng)建 StandaloneResourceManager 實(shí)例對(duì)象
- * 1、resourceManager = StandaloneResourceManager
- * 2、resourceManagerFactory = StandaloneResourceManagerFactory
- */
- resourceManager = resourceManagerFactory.createResourceManager(
- configuration, ResourceID.generate(),
- rpcService, highAvailabilityServices, heartbeatServices,
- fatalErrorHandler, new ClusterInformation(hostname, blobServer.getPort()),
- webMonitorEndpoint.getRestBaseUrl(), metricRegistry, hostname
- );
- protected ResourceManager<ResourceID> createResourceManager(
- Configuration configuration,
- ResourceID resourceId,
- RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- HeartbeatServices heartbeatServices,
- FatalErrorHandler fatalErrorHandler,
- ClusterInformation clusterInformation,
- @Nullable String webInterfaceUrl,
- ResourceManagerMetricGroup resourceManagerMetricGroup,
- ResourceManagerRuntimeServices resourceManagerRuntimeServices) {
- final Time standaloneClusterStartupPeriodTime = ConfigurationUtils.getStandaloneClusterStartupPeriodTime(configuration);
- /*************************************************
- * 注釋?zhuān)?nbsp;得到一個(gè) StandaloneResourceManager 實(shí)例對(duì)象
- */
- return new StandaloneResourceManager(
- rpcService,
- resourceId,
- highAvailabilityServices,
- heartbeatServices,
- resourceManagerRuntimeServices.getSlotManager(),
- ResourceManagerPartitionTrackerImpl::new,
- resourceManagerRuntimeServices.getJobLeaderIdService(),
- clusterInformation,
- fatalErrorHandler,
- resourceManagerMetricGroup,
- standaloneClusterStartupPeriodTime,
- AkkaUtils.getTimeoutAsTime(configuration)
- );
- }
- /**
- requestSlot():接受 solt請(qǐng)求
- sendSlotReport(..): 將solt請(qǐng)求發(fā)送TaskManager
- registerJobManager(...): 注冊(cè)job管理者。 該job指的是 提交給flink的應(yīng)用程序
- registerTaskExecutor(...): 注冊(cè)task執(zhí)行者。
- **/
- public ResourceManager(RpcService rpcService, ResourceID resourceId, HighAvailabilityServices highAvailabilityServices,
- HeartbeatServices heartbeatServices, SlotManager slotManager, ResourceManagerPartitionTrackerFactory clusterPartitionTrackerFactory,
- JobLeaderIdService jobLeaderIdService, ClusterInformation clusterInformation, FatalErrorHandler fatalErrorHandler,
- ResourceManagerMetricGroup resourceManagerMetricGroup, Time rpcTimeout) {
- /*************************************************
- * 注釋?zhuān)?nbsp;當(dāng)執(zhí)行完畢這個(gè)構(gòu)造方法的時(shí)候,會(huì)觸發(fā)調(diào)用 onStart() 方法執(zhí)行
- */
- super(rpcService, AkkaRpcServiceUtils.createRandomName(RESOURCE_MANAGER_NAME), null);
- protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
- this.rpcService = checkNotNull(rpcService, "rpcService");
- this.endpointId = checkNotNull(endpointId, "endpointId");
- /*************************************************
- * 注釋?zhuān)篟esourceManager 或者 TaskExecutor 中的 RpcServer 實(shí)現(xiàn)
- * 以 ResourceManager 為例說(shuō)明:
- * 啟動(dòng) ResourceManager 的 RPCServer 服務(wù)
- * 這里啟動(dòng)的是 ResourceManager 的 Rpc 服務(wù)端。
- * 接收 TaskManager 啟動(dòng)好了而之后, 進(jìn)行注冊(cè)和心跳,來(lái)匯報(bào) Taskmanagaer 的資源情況
- * 通過(guò)動(dòng)態(tài)代理的形式構(gòu)建了一個(gè)Server
- */
- this.rpcServer = rpcService.startServer(this);
5. 在創(chuàng)建resourceManager同級(jí):?jiǎn)?dòng)任務(wù)接收器Starting Dispatcher
- /*************************************************
- * 創(chuàng)建 并啟動(dòng) Dispatcher
- * 1、dispatcherRunner = DispatcherRunnerLeaderElectionLifecycleManager
- * 2、dispatcherRunnerFactory = DefaultDispatcherRunnerFactory
- * 第一個(gè)參數(shù):ZooKeeperLeaderElectionService
- * -
- * 老版本: 這個(gè)地方是直接創(chuàng)建一個(gè) Dispatcher 對(duì)象然后調(diào)用 dispatcher.start() 來(lái)啟動(dòng)
- * 新版本: 直接創(chuàng)建一個(gè) DispatcherRunner, 內(nèi)部就是要?jiǎng)?chuàng)建和啟動(dòng) Dispatcher
- * -
- * DispatcherRunner 是對(duì) Dispatcher 的封裝。
- * DispatcherRunner被創(chuàng)建的代碼的內(nèi)部,會(huì)創(chuàng)建 Dispatcher并啟動(dòng)
- */
- log.debug("Starting Dispatcher.");
- dispatcherRunner = dispatcherRunnerFactory.createDispatcherRunner(
- highAvailabilityServices.getDispatcherLeaderElectionService(), fatalErrorHandler,
- // TODO_ZYM 注釋?zhuān)?nbsp;注意第三個(gè)參數(shù)
- new HaServicesJobGraphStoreFactory(highAvailabilityServices),
- ioExecutor, rpcService, partialDispatcherServices
- );
Dispatcher 啟動(dòng)后,將會(huì)等待任務(wù)提交,如果有任務(wù)提交,則會(huì)經(jīng)過(guò)submitJob(...)函數(shù)進(jìn)入后續(xù)處理。
提交(一個(gè)Flink應(yīng)用的提交必須經(jīng)過(guò)三個(gè)graph的轉(zhuǎn)換)
首先看下一些名詞
StreamGraph
是根據(jù)用戶(hù)通過(guò) Stream API 編寫(xiě)的代碼生成的最初的圖。用來(lái)表示程序的拓?fù)浣Y(jié)構(gòu)??梢杂靡粋€(gè) DAG 來(lái)表示),DAG 的頂點(diǎn)是 StreamNode,邊是 StreamEdge,邊包含了由哪個(gè) StreamNode 依賴(lài)哪個(gè) StreamNode。
- StreamNode:用來(lái)代表 operator 的類(lèi),并具有所有相關(guān)的屬性,如并發(fā)度、入邊和出邊等。
- StreamEdge:表示連接兩個(gè)StreamNode的邊。
DataStream 上常見(jiàn)的 transformation 有 map、flatmap、filter等(見(jiàn)DataStream Transformation了解更多)。這些transformation會(huì)構(gòu)造出一棵 StreamTransformation 樹(shù),通過(guò)這棵樹(shù)轉(zhuǎn)換成 StreamGraph
以map方法為例,看看源碼
- public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper) {
- // 通過(guò)java reflection抽出mapper的返回值類(lèi)型
- TypeInformation<R> outType = TypeExtractor.getMapReturnTypes(clean(mapper), getType(),
- Utils.getCallLocationName(), true);
- // 返回一個(gè)新的DataStream,SteramMap 為 StreamOperator 的實(shí)現(xiàn)類(lèi)
- return transform("Map", outType, new StreamMap<>(clean(mapper)));
- }
- public <R> SingleOutputStreamOperator<R> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {
- // read the output type of the input Transform to coax out errors about MissingTypeInfo
- transformation.getOutputType();
- // 新的transformation會(huì)連接上當(dāng)前DataStream中的transformation,從而構(gòu)建成一棵樹(shù)
- OneInputTransformation<T, R> resultTransform = new OneInputTransformation<>(
- this.transformation,
- operatorName,
- operator,
- outTypeInfo,
- environment.getParallelism());
- @SuppressWarnings({ "unchecked", "rawtypes" })
- SingleOutputStreamOperator<R> returnStream = new SingleOutputStreamOperator(environment, resultTransform);
- // 所有的transformation都會(huì)存到 env 中,調(diào)用execute時(shí)遍歷該list生成StreamGraph
- getExecutionEnvironment().addOperator(resultTransform);
- return returnStream;
- }
map轉(zhuǎn)換將用戶(hù)自定義的函數(shù)MapFunction包裝到StreamMap這個(gè)Operator中,再將StreamMap包裝到OneInputTransformation,最后該transformation存到env中,當(dāng)調(diào)用env.execute時(shí),遍歷其中的transformation集合構(gòu)造出StreamGraph
JobGraph
(1) StreamGraph經(jīng)過(guò)優(yōu)化后生成了 JobGraph,提交給 JobManager 的數(shù)據(jù)結(jié)構(gòu)。主要的優(yōu)化為,將多個(gè)符合條件的節(jié)點(diǎn) chain 在一起作為一個(gè)節(jié)點(diǎn)。
- 將并不涉及到 shuffle 的算子進(jìn)行合并。
- 對(duì)于同一個(gè) operator chain 里面的多個(gè)算子,會(huì)在同一個(gè) task 中執(zhí)行。
- 對(duì)于不在同一個(gè) operator chain 里的算子,會(huì)在不同的 task 中執(zhí)行。
(2) JobGraph 用來(lái)由 JobClient 提交給 JobManager,是由頂點(diǎn)(JobVertex)、中間結(jié)果(IntermediateDataSet)和邊(JobEdge)組成的 DAG 圖。
(3) JobGraph 定義作業(yè)級(jí)別的配置,而每個(gè)頂點(diǎn)和中間結(jié)果定義具體操作和中間數(shù)據(jù)的設(shè)置。
JobVertex
JobVertex 相當(dāng)于是 JobGraph 的頂點(diǎn)。經(jīng)過(guò)優(yōu)化后符合條件的多個(gè)StreamNode可能會(huì)chain在一起生成一個(gè)JobVertex,即一個(gè)JobVertex包含一個(gè)或多個(gè)operator,JobVertex的輸入是JobEdge,輸出是IntermediateDataSet。
IntermediateDataSet
JobVertex的輸出,即經(jīng)過(guò)operator處理產(chǎn)生的數(shù)據(jù)集。
JobEdge
job graph中的一條數(shù)據(jù)傳輸通道。source 是IntermediateDataSet,sink 是 JobVertex。即數(shù)據(jù)通過(guò)JobEdge由IntermediateDataSet傳遞給目標(biāo)JobVertex。
(1) 首先是通過(guò)API會(huì)生成transformations,通過(guò)transformations會(huì)生成StreamGraph。
(2)將StreamGraph的某些StreamNode Chain在一起生成JobGraph,前兩步轉(zhuǎn)換都是在客戶(hù)端完成。
(3)最后會(huì)將JobGraph轉(zhuǎn)換為ExecutionGraph,相比JobGraph會(huì)增加并行度的概念,這一步是在Jobmanager里完成。
ExecutionJobVertex
ExecutionJobVertex一一對(duì)應(yīng)JobGraph中的JobVertex
ExecutionVertex
一個(gè)ExecutionJobVertex對(duì)應(yīng)n個(gè)ExecutionVertex,其中n就是算子的并行度。ExecutionVertex就是并行任務(wù)的一個(gè)子任務(wù)
Execution
Execution 是對(duì) ExecutionVertex 的一次執(zhí)行,通過(guò) ExecutionAttemptId 來(lái)唯一標(biāo)識(shí)。
IntermediateResult
在 JobGraph 中用 IntermediateDataSet 表示 JobVertex 的對(duì)外輸出,一個(gè) JobGraph 可能有 n(n >=0) 個(gè)輸出。在 ExecutionGraph 中,與此對(duì)應(yīng)的就是 IntermediateResult。每一個(gè) IntermediateResult 就有 numParallelProducers(并行度) 個(gè)生產(chǎn)者,每個(gè)生產(chǎn)者的在相應(yīng)的 IntermediateResult 上的輸出對(duì)應(yīng)一個(gè) IntermediateResultPartition。IntermediateResultPartition 表示的是 ExecutionVertex 的一個(gè)輸出分區(qū)
ExecutionEdge
ExecutionEdge 表示 ExecutionVertex 的輸入,通過(guò) ExecutionEdge 將 ExecutionVertex 和 IntermediateResultPartition 連接起來(lái),進(jìn)而在不同的 ExecutionVertex 之間建立聯(lián)系。
ExecutionGraph的構(gòu)建
- 構(gòu)建JobInformation
- 構(gòu)建ExecutionGraph
- 將JobGraph進(jìn)行拓?fù)渑判?獲取sortedTopology頂點(diǎn)集合
- // ExecutionGraphBuilder
- public static ExecutionGraph buildGraph(
- @Nullable ExecutionGraph prior,
- JobGraph jobGraph,
- ...) throws JobExecutionException, JobException {
- // 構(gòu)建JobInformation
- // 構(gòu)建ExecutionGraph
- // 將JobGraph進(jìn)行拓?fù)渑判?獲取sortedTopology頂點(diǎn)集合
- List<JobVertex> sortedTopology = jobGraph.getVerticesSortedTopologicallyFromSources();
- executionGraph.attachJobGraph(sortedTopology);
- return executionGraph;
- }
構(gòu)建ExecutionJobVertex,連接IntermediateResultPartition和ExecutionVertex
- //ExecutionGraph
- public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
- for (JobVertex jobVertex : topologiallySorted) {
- // 構(gòu)建ExecutionJobVertex
- ExecutionJobVertex ejv = new ExecutionJobVertex(
- this,
- jobVertex,
- 1,
- maxPriorAttemptsHistoryLength,
- rpcTimeout,
- globalModVersion,
- createTimestamp);
- // 連接IntermediateResultPartition和ExecutionVertex
- ev.connectToPredecessors(this.intermediateResults);
- }
- // ExecutionJobVertex
- public void connectToPredecessors(Map<IntermediateDataSetID, IntermediateResult> intermediateDataSets) throws JobException {
- List<JobEdge> inputs = jobVertex.getInputs();
- for (int num = 0; num < inputs.size(); num++) {
- JobEdge edge = inputs.get(num);
- IntermediateResult ires = intermediateDataSets.get(edge.getSourceId());
- this.inputs.add(ires);
- int consumerIndex = ires.registerConsumer();
- for (int i = 0; i < parallelism; i++) {
- ExecutionVertex ev = taskVertices[i];
- ev.connectSource(num, ires, edge, consumerIndex);
- }
- }
- }
拆分計(jì)劃(可執(zhí)行能力)
- // ExecutionVertex
- public void connectSource(int inputNumber, IntermediateResult source, JobEdge edge, int consumerNumber) {
- final DistributionPattern pattern = edge.getDistributionPattern();
- final IntermediateResultPartition[] sourcePartitions = source.getPartitions();
- ExecutionEdge[] edges;
- switch (pattern) {
- // 下游 JobVertex 的輸入 partition 算法,如果是 forward 或 rescale 的話為 POINTWISE
- case POINTWISE:
- edges = connectPointwise(sourcePartitions, inputNumber);
- break;
- // 每一個(gè)并行的ExecutionVertex節(jié)點(diǎn)都會(huì)鏈接到源節(jié)點(diǎn)產(chǎn)生的所有中間結(jié)果IntermediateResultPartition
- case ALL_TO_ALL:
- edges = connectAllToAll(sourcePartitions, inputNumber);
- break;
- default:
- throw new RuntimeException("Unrecognized distribution pattern.");
- }
- inputEdges[inputNumber] = edges;
- for (ExecutionEdge ee : edges) {
- ee.getSource().addConsumer(ee, consumerNumber);
- }
- }
- private ExecutionEdge[] connectPointwise(IntermediateResultPartition[] sourcePartitions, int inputNumber) {
- final int numSources = sourcePartitions.length;
- final int parallelism = getTotalNumberOfParallelSubtasks();
- // 如果并發(fā)數(shù)等于partition數(shù),則一對(duì)一進(jìn)行連接
- if (numSources == parallelism) {
- return new ExecutionEdge[] { new ExecutionEdge(sourcePartitions[subTaskIndex], this, inputNumber) };
- }
- // 如果并發(fā)數(shù)大于partition數(shù),則一對(duì)多進(jìn)行連接
- else if (numSources < parallelism) {
- int sourcePartition;
- if (parallelism % numSources == 0) {
- int factor = parallelism / numSources;
- sourcePartition = subTaskIndex / factor;
- }
- else {
- float factor = ((float) parallelism) / numSources;
- sourcePartition = (int) (subTaskIndex / factor);
- }
- return new ExecutionEdge[] { new ExecutionEdge(sourcePartitions[sourcePartition], this, inputNumber) };
- }
- // 果并發(fā)數(shù)小于partition數(shù),則多對(duì)一進(jìn)行連接
- else {
- if (numSources % parallelism == 0) {
- int factor = numSources / parallelism;
- int startIndex = subTaskIndex * factor;
- ExecutionEdge[] edges = new ExecutionEdge[factor];
- for (int i = 0; i < factor; i++) {
- edges[i] = new ExecutionEdge(sourcePartitions[startIndex + i], this, inputNumber);
- }
- return edges;
- }
- else {
- float factor = ((float) numSources) / parallelism;
- int start = (int) (subTaskIndex * factor);
- int end = (subTaskIndex == getTotalNumberOfParallelSubtasks() - 1) ?
- sourcePartitions.length :
- (int) ((subTaskIndex + 1) * factor);
- ExecutionEdge[] edges = new ExecutionEdge[end - start];
- for (int i = 0; i < edges.length; i++) {
- edges[i] = new ExecutionEdge(sourcePartitions[start + i], this, inputNumber);
- }
- return edges;
- }
- }
- }
- private ExecutionEdge[] connectAllToAll(IntermediateResultPartition[] sourcePartitions, int inputNumber) {
- ExecutionEdge[] edges = new ExecutionEdge[sourcePartitions.length];
- for (int i = 0; i < sourcePartitions.length; i++) {
- IntermediateResultPartition irp = sourcePartitions[i];
- edges[i] = new ExecutionEdge(irp, this, inputNumber);
- }
- return edges;
- }
返回ExecutionGraph
TaskManager
TaskManager啟動(dòng)
- public static void runTaskManager(Configuration configuration, ResourceID resourceId) throws Exception {
- //主要初始化一堆的service,并新建一個(gè)org.apache.flink.runtime.taskexecutor.TaskExecutor
- final TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration,resourceId);
- //調(diào)用TaskExecutor的start()方法
- taskManagerRunner.start();
- }
TaskExecutor :submitTask()
接著的重要函數(shù)是shumitTask()函數(shù),該函數(shù)會(huì)通過(guò)AKKA機(jī)制,向TaskManager發(fā)出一個(gè)submitTask的消息請(qǐng)求,TaskManager收到消息請(qǐng)求后,會(huì)執(zhí)行submitTask()方法。(省略了部分代碼)。
- public CompletableFuture<Acknowledge> submitTask(
- TaskDeploymentDescriptor tdd,
- JobMasterId jobMasterId,
- Time timeout) {
- jobInformation = tdd.getSerializedJobInformation().deserializeValue(getClass().getClassLoader());
- taskInformation = tdd.getSerializedTaskInformation().deserializeValue(getClass().getClassLoader());
- TaskMetricGroup taskMetricGroup = taskManagerMetricGroup.addTaskForJob(xxx);
- InputSplitProvider inputSplitProvider = new RpcInputSplitProvider(xxx);
- TaskManagerActions taskManagerActions = jobManagerConnection.getTaskManagerActions();
- CheckpointResponder checkpointResponder = jobManagerConnection.getCheckpointResponder();
- LibraryCacheManager libraryCache = jobManagerConnection.getLibraryCacheManager();
- ResultPartitionConsumableNotifier resultPartitionConsumableNotifier = jobManagerConnection.getResultPartitionConsumableNotifier();
- PartitionProducerStateChecker partitionStateChecker = jobManagerConnection.getPartitionStateChecker();
- final TaskLocalStateStore localStateStore = localStateStoresManager.localStateStoreForSubtask(
- jobId,
- tdd.getAllocationId(),
- taskInformation.getJobVertexId(),
- tdd.getSubtaskIndex());
- final JobManagerTaskRestore taskRestore = tdd.getTaskRestore();
- final TaskStateManager taskStateManager = new TaskStateManagerImpl(
- jobId,
- tdd.getExecutionAttemptId(),
- localStateStore,
- taskRestore,
- checkpointResponder);
- //新建一個(gè)Task
- Task task = new Task(xxxx);
- log.info("Received task {}.", task.getTaskInfo().getTaskNameWithSubtasks());
- boolean taskAdded;
- try {
- taskAdded = taskSlotTable.addTask(task);
- } catch (SlotNotFoundException | SlotNotActiveException e) {
- throw new TaskSubmissionException("Could not submit task.", e);
- }
- if (taskAdded) {
- //啟動(dòng)任務(wù)
- task.startTaskThread();
- return CompletableFuture.completedFuture(Acknowledge.get());
- }
最后創(chuàng)建執(zhí)行Task的線程,然后調(diào)用startTaskThread()來(lái)啟動(dòng)具體的執(zhí)行線程,Task線程內(nèi)部的run()方法承載了被執(zhí)行的核心邏輯。
Task是執(zhí)行在TaskExecutor進(jìn)程里的一個(gè)線程,下面來(lái)看看其run方法
(1) 檢測(cè)當(dāng)前狀態(tài),正常情況為CREATED,如果是FAILED或CANCELING直接返回,其余狀態(tài)將拋異常。
(2) 讀取DistributedCache文件。
(3) 啟動(dòng)ResultPartitionWriter和InputGate。
(4) 向taskEventDispatcher注冊(cè)partitionWriter。
(5) 根據(jù)nameOfInvokableClass加載對(duì)應(yīng)的類(lèi)并實(shí)例化。
(6) 將狀態(tài)置為RUNNING并執(zhí)行invoke方法。
- public void run() {
- while (true) {
- ExecutionState current = this.executionState;
- invokable = loadAndInstantiateInvokable(userCodeClassLoader, nameOfInvokableClass);
- network.registerTask(this);
- Environment env = new RuntimeEnvironment(. . . . );
- invokable.setEnvironment(env);
- // actual task core work
- if (!transitionState(ExecutionState.DEPLOYING, ExecutionState.RUNNING)) {
- }
- // notify everyone that we switched to running
- notifyObservers(ExecutionState.RUNNING, null);
- executingThread.setContextClassLoader(userCodeClassLoader);
- // run the invokable
- invokable.invoke();
- if (transitionState(ExecutionState.RUNNING, ExecutionState.FINISHED)) {
- notifyObservers(ExecutionState.FINISHED, null);
- }
- Finally{
- // free the network resources
- network.unregisterTask(this);
- // free memory resources
- if (invokable != null) {
- memoryManager.releaseAll(invokable);
- }
- libraryCache.unregisterTask(jobId, executionId);
- removeCachedFiles(distributedCacheEntries, fileCache);
總結(jié)
整體的流程與架構(gòu)可能三兩張圖或者三言?xún)烧Z(yǔ)就可以勾勒出畫(huà)面,但是背后源碼的實(shí)現(xiàn)是艱辛的。源碼的復(fù)雜度和當(dāng)初設(shè)計(jì)框架的抓狂感,我們只有想象?,F(xiàn)在我們只是站在巨人的肩膀上去學(xué)習(xí)。
本篇的主題是"Flink架構(gòu)與執(zhí)行流程",做下小結(jié),F(xiàn)link on Yarn的提交執(zhí)行流程:
1 Flink任務(wù)提交后,Client向HDFS上傳Flink的Jar包和配置。
2 向Yarn ResourceManager提交任務(wù)。
3 ResourceManager分配Container資源并通知對(duì)應(yīng)的NodeManager啟動(dòng)ApplicationMaster。
4 ApplicationMaster啟動(dòng)后加載Flink的Jar包和配置構(gòu)建環(huán)境。
5 啟動(dòng)JobManager之后ApplicationMaster向ResourceManager申請(qǐng)資源啟動(dòng)TaskManager。
6 ResourceManager分配Container資源后,由ApplicationMaster通知資源所在節(jié)點(diǎn)的NodeManager啟動(dòng)TaskManager。
7 NodeManager加載Flink的Jar包和配置構(gòu)建環(huán)境并啟動(dòng)TaskManager。
8 TaskManager啟動(dòng)后向JobManager發(fā)送心跳包,并等待JobManager向其分配任務(wù)。