近年来,基于hadoop的sql框架层出不穷,presto也是其中的一员.从2012年发展至今,依然保持年轻的活力(版本迭代依然很快),presto的相关介绍,咱们就不赘述了,相信看官多对presto有或多或少的了解,详细的一些说明能够看官网(https://prestodb.io)的说明.java
presto自身功能和思想富有先进性,虽然因为是内存计算,稳定性方面还有很大提高空间,但总体依然在adhoc方面有很好的竞争力.其中在catalog加载的方式上来讲比较的固化,官方并无作出动态的方案出来,致使在添加catalog后必须重启整个集群才能够将新添加的catalog数据源添加到presto中,这无疑在实际的生产环境中很不友好.尤为是在一些中台项目中,须要动态规划的东西很是多.这种模式的catalog添加方式显然不能知足咱们的开发须要.mysql
所以,在环境的加持下,对presto的加载catlog的方式的源码进行了改造,使其具备热动态添加的功能.咱们采用了外部数据库做为他的catlog资源库,对其进行热加载sql
(1)添加restful API请求接口.数据库
为了使框架自己具备添加catalog的功能,须要使其自己具备Api访问接口的方式来来对catalog的资源进行调整的功能api
1.新增CatalogResource类来实现api的请求接口
2.新增TimiCatalogStoreConfig类来实现与数据库交互的持久层
3.新增TimiCatalogStore类来替换本来的catlog加载类
4.新增CatalogInfo类来实现对catalog Model信息的解析
#1 CatalogResource
@Path("/presto/catalog")
public class CatalogResource{
@GET
@Path("test")
public Response test()
{
return Response.ok("Hello world").build();
}
}
在ServerMainModule类中setup方法,最后一行添加jaxrsBinder(binder).bind(CatalogResource.class);将添加的请求类添加进来,而后启动主服务,并确认所开启的presto的请求接口地址,默认端口是:8080请求http://localhost:8080/presto/catalog/test服务器
返回 "Hello world" 则表示restful API 接口添加成功.restful
#2 TimiCatalogStoreConfig 类中主要实现了读取数据库链接配置,以及具体执行的catalog执行动做,并使用jaxrsBinder(binder).bind(TimiCatalogStoreConfig.class);注入到项目启动的容器中.并将Announcer,disabledCatalogs,ConnectorManager注入到类中.具体实现框架
public class TimiCatalogStoreConfig {
private final Announcer announcer;
private static final Logger log = Logger.get(TimiCatalogStoreConfig.class);
private final Set<String> disabledCatalogs;
private final ConnectorManager connectorManager;
public TimiCatalogStoreConfig(Announcer announcer,Set<String> disabledCatalogs,ConnectorManager connectorManager ) {
this.announcer = announcer;
this.disabledCatalogs = ImmutableSet.copyOf(disabledCatalogs);
this.connectorManager = connectorManager;
}
}
而后就是实现对catlog增删查改动做,并将操做的结构实现到ConnectorManager中,ide
首先将Server中的CatalogStore替换成咱们自定义实现的TimiCatalogStore并注入相关类oop
@Inject
public TimiCatalogStore(ConnectorManager connectorManager, Announcer announcer, StaticCatalogStoreConfig config,TimiCatalogStoreConfig catalogStoreConfig) {
this(connectorManager,
announcer,
config.getCatalogConfigurationDir(),
firstNonNull(config.getDisabledCatalogs(), ImmutableList.of()),
catalogStoreConfig
);
}
而后实现loadCatalogs方法,首次调用的时候使用load();方法加载mysql中存储的全部catlog,而后使用ScheduledExecutorService定时的方式从mysql中提取有变化的catlog加载到presto的ConnectorManager中.
public static void updateConnectorIdAnnouncement(Announcer announcer, CatalogName connectorId)
{
//
// This code was copied from PrestoServer, and is a hack that should be removed when the connectorId property is removed
//
// get existing announcement
ServiceAnnouncement announcement = getPrestoAnnouncement(announcer.getServiceAnnouncements());
// update connectorIds property
Map<String, String> properties = new LinkedHashMap<>(announcement.getProperties());
String property = nullToEmpty(properties.get("connectorIds"));
Set<String> connectorIds = new LinkedHashSet<>(Splitter.on(',').trimResults().omitEmptyStrings().splitToList(property));
connectorIds.add(connectorId.toString());
properties.put("connectorIds", Joiner.on(',').join(connectorIds));
// update announcement
announcer.removeServiceAnnouncement(announcement.getId());
announcer.addServiceAnnouncement(serviceAnnouncement(announcement.getType()).addProperties(properties).build());
announcer.forceAnnounce();
}
在这里咱们设定的1分钟从mysql库充更新一次catalog列表
scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
reload();
}
}, 60, 60, TimeUnit.SECONDS);
调用reload方法定时读取
public void reload() {
try{
//获取最新的catalogs
Map<String, CatalogInfo> catalogInfos = catalogStoreConfig.load();
catalogInfos.forEach(
(key, catalogInfo) -> {
if (!catalogInfoMap.containsKey(key)) {
//相同--catlog
try {
System.out.println("添加数据源"+JSON.toJSONString(catalogInfo));
// log.info("添加数据源:{}",JSON.toJSONString(catalogInfos.get(key)));
CatalogName catalogName = loadCatalog(catalogInfo);
updateConnectorIdAnnouncement(announcer,catalogName);
} catch (Exception e) {
e.printStackTrace();
}
} else {
//不一样catlog
if (!JSON.toJSONString(catalogInfoMap.get(key)).equals(JSON.toJSONString(catalogInfo))){
connectorManager.dropConnection(catalogInfo.getCatalogName());
try {
System.out.println("添加数据源"+JSON.toJSONString(catalogInfo));
CatalogName catalogName = loadCatalog(catalogInfo);
updateConnectorIdAnnouncement(announcer,catalogName);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
);
catalogInfoMap.putAll(catalogInfos);
}catch (Exception e){
e.printStackTrace();
}
}
从mysql库中取出来的catlog信息和对现有的catlog进行对比,若是是不一样的catlog就添加到presto中,重复的catlog不添加,删除的catlog就从现有的catlog管理器中删除,以此来达到动态添加catlog的动做,不须要重启presto服务器.