1.概述

转载： Flink1.10基于工厂模式的任务提交与SPI机制

Flink任务执行模式包含了yarn-session、standalone、per-job、local, 在1.10中又增加k8s的执行模式，那么在任务提交过程中如何根据不同的执行模式进行任务提交呢？

主要通过两个接口来实现：PipelineExecutorFactory 与 PipelineExecutor。PipelineExecutorFactory用于在不同模式下创建不同的PipelineExecutor, 用于提交任务，PipelineExecutorFactory表示的一个创建执行器工厂接口，PipelineExecutor 表示一个执行器接口，正如你所想这里使用的就是经典的工厂设计模式，在任务提交过程中会根据不同的提交模式，使用不同的PipelineExecutorFactory创建不同的PipelineExecutor。

public interface PipelineExecutorFactory {/*** Returns the name of the executor that this factory creates.*/String getName();/**根据configuration判断是否满足当前的factory*/boolean isCompatibleWith(final Configuration configuration);/*** 获取对应模式下的executor*/PipelineExecutor getExecutor(final Configuration configuration);
}

PipelineExecutorFactory几个实现分别为：

LocalExecutorFactory(local)
RemoteExecutorFactory(standalone)
YarnJobClusterExecutorFactory(per-job)
YarnSessionClusterExecutorFactory(yarn-session)

public interface PipelineExecutor {/*** 执行任务*/CompletableFuture<JobClient> execute(final Pipeline pipeline, final Configuration configuration) throws Exception;
}

PipelineExecutor对应实现:

RemoteExecutor(standalone)
LocalExecutor(local)
YarnJobClusterExecutor(per-job)
YarnSessionClusterExecutor(yarn-session)

那么具体是如何选择factory呢？由PipelineExecutorServiceLoader接口来完成，其只有一个实现类DefaultExecutorServiceLoader，透过命名你可能会才想到这里面用到了ServiceLoader，你的猜想是正确的，它就是通过SPI机制去加载flink所提供的不同factory，在META-INF.services 下可以找到其对应的配置：

DefaultExecutorServiceLoader.java部分源码

//SPI机制
private static final ServiceLoader<PipelineExecutorFactory> defaultLoader = ServiceLoader.load(PipelineExecutorFactory.class);//获取对应的factory
public PipelineExecutorFactory getExecutorFactory(final Configuration configuration) {checkNotNull(configuration);final List<PipelineExecutorFactory> compatibleFactories = new ArrayList<>();final Iterator<PipelineExecutorFactory> factories = defaultLoader.iterator();while (factories.hasNext()) {try {final PipelineExecutorFactory factory = factories.next();//判断标准 根据任务启动配置if (factory != null && factory.isCompatibleWith(configuration)) {compatibleFactories.add(factory);}} catch (Throwable e) {if (e.getCause() instanceof NoClassDefFoundError) {LOG.info("Could not load factory due to missing dependencies.");} else {throw e;}}}//只能有一个factory符合要求if (compatibleFactories.size() > 1) {final String configStr =configuration.toMap().entrySet().stream().map(e -> e.getKey() + "=" + e.getValue()).collect(Collectors.joining("\n"));throw new IllegalStateException("Multiple compatible client factories found for:\n" + configStr + ".");}return compatibleFactories.isEmpty() ? null : compatibleFactories.get(0);
}

ServiceLoader.load(PipelineExecutorFactory.class) 会从类路径的META-INF.services下找到PipelineExecutorFactory的全路径文件，然后实例化出所有的factory，通过PipelineExecutorFactory.isCompatibleWith找到匹配的factory。

此处的PipelineExecutor体系参考：【Flink】Flink PipelineExecutor 体系

【Flink】FLink PipelineExecutorFactory 基于工厂模式的任务提交与SPI机制相关推荐

php实现工厂模式,PHP基于工厂模式实现的计算器实例
本文实例讲述了PHP基于工厂模式实现的计算器.分享给大家供大家参考.具体如下: abstract class Calculator { private $number1; private $numbe ...
Python 设计模式之工厂模式
工厂模式是一个在软件开发中用来创建对象的设计模式. 工厂模式包涵一个超类.这个超类提供一个抽象化的接口来创建一个特定类型的对象,而不是决定哪个对象可以被创建. 为了实现此方法,需要创建一个工厂类并返回 ...
二、工厂模式——在工厂里能找到你的对象
文章目录工厂模式 1. 简单工厂 2. 缓存方式 3. 工厂模式 4. 工厂的工厂 5. 抽象工厂 6. 工厂模式与IOC容器总结设计模式是面向问题.场景而总结产生的设计思路.是解决问题的套路. ...
策略模式+工厂模式（反射）+枚举代替大量 if..else if..
实际项目中我们经常碰到需要使用if-else-if的分支判断这种情况. 这种写法带来一些弊端. 一旦分支多太多,逻辑复杂,会导致代码十分冗长,增加阅读难度. 如果需要增加或减少分支,需要改动if-el ...
浅谈Spring框架应用的设计模式（一）——工厂模式
文章目录前言一.工厂模式介绍 1.简单工厂模式 (1)静态工厂模式 (2)利用反射机制实现的简单工厂 2.工厂方法模式 3.抽象工厂模式二.Spring框架中工厂模式的重要应用 1.BeanFa ...
【flink】Flink 1.12.2 源码浅析 : yarn-per-job模式解析 yarn 提交过程解析
1.概述转载:Flink 1.12.2 源码浅析 : yarn-per-job模式解析 [二] 请大家看原文去. 接上文Flink 1.12.2 源码分析 : yarn-per-job模式浅析 [一 ...
2021年大数据Flink（三）：Flink安装部署 Local本地模式
目录 Flink安装部署 Local本地模式原理操作测试 Flink安装部署 Flink支持多种安装模式 - Local-本地单机模式,学习测试时使用 - Standalone-独立集群模式,F ...
Flink运行时架构及各部署模式下作业提交流程
1.运行时架构 1.1 核心组件 1.1.1 JobManager 作业管理器,对于一个提交执行的作业,JobManager 是真正意义上的"管理者"(Master),负责管理调度 ...
【设计模式-手写源码-附1】-简单工厂模式-基于魔兽争霸冰封王座
1:主题拆解 ①依赖倒置原则-SimpleFactory ②简单工厂+ 配置文件=可配置 ③简单工厂+ 配置文件+反射=可配置可扩展 ④简单工厂升级IOC控制反转 2:基本介绍 ①学习设计模式的套路: ...

【Flink】FLink PipelineExecutorFactory 基于工厂模式的任务提交与SPI机制

1.概述

【Flink】FLink PipelineExecutorFactory 基于工厂模式的任务提交与SPI机制相关推荐

最新文章

热门文章