Presto多协调者实现

写在前面：
Presto 是facebook开发的一款开源大数据OLAP引擎，基于内存计算和作业分发，直接使用sql分析，不需要数据预处理，类似于impala和sparksql，而且它本身不保存任何数据，通过连接数据源取数。

在Presto的架构中，主要有几种角色

Coordinator：
Coordinator服务器是用来解析语句，执行计划分析和管理Presto的worker结点。Presto安装必须有一个Coordinator和多个worker。
Coordinator跟踪每个work的活动情况并协调查询语句的执行。 Coordinator为每个查询建立模型，模型包含多个stage，每个stage再转为task分发到不同的worker上执行。Coordinator再做结果的聚合。

Worker：
Worker是负责执行任务和处理数据。

Discovery Server：
通常内嵌于coordinator节点中，也可以单独部署，用于Worker的注册，让Coordinator知道Worker有多少，地址是啥

问题：
前面提到，presto集群是基于内存和作业并行处理来执行分析任务，对于机器不多的单个集群来讲，并发很低，不过这倒是次要的，很多情况下，会出现cpu和内存打满的情形，这就直接影响到了可用性（最主要的原因是穷，机器的硬件都是低配且机器只有三台），而最为明显的瓶颈就首先出现在了Coordinator上，多条查询语句并发执行时，有一些直接被阻塞了，而且内存直接打满，通过jstat命令查看，GC就没停下来过，但是Worker的话，都还好

屏幕截图 2021-03-31 162921.jpg

开始我的设想是，增加一台协调者，使用集群的Discovery Server，然后使用nginx做负载，将请求分流到两台协调者上，以减轻Coordinator压力同时增强并发能力

nginx.conf

   upstream backend {
      server 10.100.218.110:8090;
      server 10.101.126.93:8090;
    }


    server {
        listen       8060;
        server_name  tt;

        location / {
            root   html;
            index  index.html index.htm;
            proxy_pass http://backend;
            proxy_redirect off;
            proxy_set_header Host $http_host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        }

        error_page   500 502 503 504  /50x.html;
        location = /50x.html {
            root   html;
        }

    }

Coordinator-1(最开始的协调者) config:

coordinator=true
node-scheduler.include-coordinator=true
http-server.http.port=8090
discovery-server.enabled=true
discovery.uri=http://10.100.218.110:8090
query.max-memory-per-node=6GB
query.max-total-memory-per-node=8GB
query.low-memory-killer.policy=total-reservation
http-server.max-request-header-size=32MB
http-server.max-response-header-size=32MB
task.concurrency=16

Coordinator-2(新加入的协调者) config:

coordinator=true
node-scheduler.include-coordinator=false
http-server.http.port=8090
discovery-server.enabled=false
discovery.uri=http://10.100.218.110:8090
query.max-memory-per-node=4GB
query.max-total-memory-per-node=5GB
query.low-memory-killer.policy=total-reservation
http-server.max-request-header-size=32MB
http-server.max-response-header-size=32MB
task.concurrency=16

但是，出问题了，所有查询都失败，原因是presto-jdbc执行查询请求并不只是发一个http请求就完事儿，而是会有多个请求去发送预处理，赋值，查询队列状况，获取执行结果等等。通过Fiddler抓包的结果也验证了这一点

企业微信截图_16171750768073.png

要想解决这个问题就必须将这些请求路由到同一个Coordinator节点上，我预先设想的是在presto-jdbc驱动发送请求的时候拦截下来，然后在Header里面强行塞线程id，然后nginx再根据每个请求的线程id来求模做路由

后面发现，在应用端做路由也是可行的,就是注入两个数据源，获取连接的时候根据线程id求模来决定走哪个连接

PrestoRouteDataSource （多数据源实现）

import com.zaxxer.hikari.HikariDataSource;

import javax.sql.DataSource;
import java.io.PrintWriter;
import java.sql.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.logging.Logger;


public class PrestoRouteDataSource implements DataSource {


    private CopyOnWriteArrayList<DataSource> corAddr = new CopyOnWriteArrayList<DataSource>();


    public PrestoRouteDataSource(String addr) {

            String[] split = addr.split(";");
            for (String s : split) {
                HikariDataSource hikariDataSource = new HikariDataSource();
                hikariDataSource.setJdbcUrl(s);
                hikariDataSource.setUsername("root");
                hikariDataSource.setDriverClassName("com.facebook.presto.jdbc.PrestoDriver");
                hikariDataSource.setPoolName("presto:" + s);
                corAddr.add(hikariDataSource);
            }

    }

    private DataSource getPrestoRouteDataSource(){
        long sessionId =  Thread.currentThread().getId();
        String as = String.valueOf(sessionId);
        int subIndex = (as.length()>=4)?(as.length()-3):(as.length()-1);
        int delSessionId =  Integer.parseInt(as.substring(subIndex));
        int index = delSessionId % corAddr.size() ;
        DataSource dataSource = corAddr.get(index);
        return dataSource ;
    }

    @Override
    public Connection getConnection() throws SQLException {


        return getPrestoRouteDataSource().getConnection();
    }

    @Override
    public Connection getConnection(String username, String password) throws SQLException {
        return getPrestoRouteDataSource().getConnection();
    }

    @Override
    public <T> T unwrap(Class<T> iface) throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isWrapperFor(Class<?> iface) throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public PrintWriter getLogWriter() throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setLogWriter(PrintWriter out) throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void setLoginTimeout(int seconds) throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public int getLoginTimeout() throws SQLException {
        throw new UnsupportedOperationException();
    }

    @Override
    public Logger getParentLogger() throws SQLFeatureNotSupportedException {
        throw new UnsupportedOperationException();
    }
}

数据源配置

import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;

import javax.sql.DataSource;
import java.io.IOException;
import java.util.*;


@Configuration
@MapperScan(basePackages = {"lvye.java.datamiddle.dao.presto"}, sqlSessionFactoryRef = "prestoSqlSessionFactory")
public class PerstoDataSourceConfig {

    @Value("${presto.jdbc-url}")
    private String url;

    @Bean(name = "prestoDataSource")
    public DataSource prestoDataSource() {
        return new PrestoRouteDataSource(url);

    }

    @Bean(name = "prestoSqlSessionFactory")
    public SqlSessionFactory mallSqlSessionFactory(@Qualifier("prestoDataSource") DataSource mallDataSource)
            throws Exception {
        final SqlSessionFactoryBean sessionFactory = new SqlSessionFactoryBean();
        sessionFactory.setTypeAliasesPackage("lvye.java.datamiddle.model.entity");
        sessionFactory.setDataSource(mallDataSource);
        sessionFactory.setMapperLocations(resolveMapperLocations());
        try {
            //开启驼峰命名转换
            Objects.requireNonNull(sessionFactory.getObject()).getConfiguration().setMapUnderscoreToCamelCase(true);
            sessionFactory.getObject().getConfiguration().setJdbcTypeForNull(null);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
        return sessionFactory.getObject();
    }

    private Resource[] resolveMapperLocations() {
        ResourcePatternResolver resourceResolver = new PathMatchingResourcePatternResolver();
        List<String> mapperLocations = new ArrayList<>();
        mapperLocations.add("classpath*:/mapper/presto/*.xml");
        List<Resource> resources = new ArrayList<>();
        for (String mapperLocation : mapperLocations) {
            try {
                Resource[] mappers = resourceResolver.getResources(mapperLocation);
                resources.addAll(Arrays.asList(mappers));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return resources.toArray(new Resource[0]);
    }


    @Bean(name = "prestoTransactionManager")
    public DataSourceTransactionManager mallTransactionManager(@Qualifier("prestoDataSource") DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }

    @Bean(name = "prestoTemplate")
    public SqlSessionTemplate prestoJdbcTemplate(@Qualifier("prestoSqlSessionFactory") SqlSessionFactory sqlSessionFactory) {
        return new SqlSessionTemplate(sqlSessionFactory);
    }

}

application.properties

presto.jdbc-url=jdbc:presto://10.100.218.110:8090/kudu/tclydm;jdbc:presto://10.101.126.93:8090/kudu/tclydm

测试1（多协调者）

multic.jpg

测试2（单协调者）

single.jpg

性能对比：
（多协调者）总耗时 37.66s 平均耗时：3.13s 最大耗时：6.67s 最小耗时：764ms

（单协调者）总耗时 131.01s 平均耗时：10.91s 最大耗时：15.78s 最小耗时：1.6s

写在后面:presto官方显然也意识到了这个问题，多协调者集群目前加入了路线图，目前版本还没有发布