Presto是一个Facebook开源的分布式SQL查询引擎，适用于交互式分析查询，数据量支持GB到PB字节。客户在使用Presto的时候发现单个集群不能满足业务需求，而建立多个Presto集群之后，如何在集群间调度任务就成为一个问题。在Presto中，一个Query执行周期内需要客户端和服务端进行多次的HTTP请求，在多集群模式下，如何保证同一个Query的请求都分发到同一个集群呢？
Lyft为解决这个问题而开发了presto-gateway这个工具，并将它开源出来。（https://github.com/lyft/presto-gateway）
Presto-gateway通过负载将不同的请求负载到不同的presto集群。
搭建流程这里，可以直接参考github，有详细的英文介绍。
https://github.com/lyft/presto-gateway
大致就是把github的项目下下来，数据库脚本执行下，修改下数据库配置文件，然后打包，运行即可。按照文档来就行。
主要修改gateway-ha/gateway-ha-config.yml就够了。这里我给出来我自己配置的样例：
routingRules:
rulesEngineEnabled: False #是否开启routing_rules，如果开启，转发的时候按照规则转
# rulesConfigPath: "src/main/resources/rules/routing_rules.yml"
requestRouter:
port: 8080 #gateway启动端口
name: prestoRouter
historySize: 1000 #请求历史日志最大保留1000条
ssl: true #gateway开启https，注意，当你要转到的presto集群开了https，这里也要开https，否则请求报错。
keystorePath: /opt/presto/etc/prestokeystore.jks #这个jks建议使用域名jks，确保可以访问到开启https的不同的presto集群
keystorePass: test.com
forwardKeystore: true
#数据库配置，存放presto-gateway的一些配置表，需要提前建好。
dataStore:
jdbcUrl: jdbc:mysql://ip:host/presto?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true
user: myuser
password: xxxx
driver: com.mysql.cj.jdbc.Driver
server:
applicationConnectors:
- type: http
port: 8090
adminConnectors:
- type: http
port: 8091
#presto支持发邮件提醒presto集群状态的功能，同时你可以改源码优化提示信息。
notifier:
smtpHost: yourmailhost
smtpPort: -1
smtpAuthEnabled: true
smtpUser: hz@xxx.com
smtpPassword: 'password'
sender: hz@xxx.com
recipients:
- xiaoli@xxx.com
# This can be adjusted based on the coordinator state
monitor:
connectionTimeout: 15
modules:
- com.lyft.data.gateway.ha.module.HaGatewayProviderModule
- com.lyft.data.gateway.ha.module.ClusterStateListenerModule
managedApps:
- com.lyft.data.gateway.ha.GatewayManagedApp
- com.lyft.data.gateway.ha.clustermonitor.ActiveClusterMonitor
# Logging settings.
logging:
# The default level of all loggers. Can be OFF, ERROR, WARN, INFO, DEBUG, TRACE, or ALL.
level: INFO
# Logger-specific levels.
loggers:
com.lyft: DEBUG
appenders:
- type: console
- type: file
currentLogFilename: /var/log/prestoproxy/prestoproxy-java.log
archivedLogFilenamePattern: /var/log/prestoproxy/prestoproxy-java-%d{yyyy-MM-dd}-%i.log.gz
archivedFileCount: 7
timeZone: UTC
maxFileSize: 100MB
上面提到的域名jks，如果是用来测试或者内部用，可以通过openssl生成。在生成的前面，还需要准备一个文件，暂且命名example.com.conf。这个文件内部指定你想要配置的证书适用的域名。
[ req ] default_bits = 2048 default_keyfile = privkey.pem distinguished_name = req_distinguished_name # 生成v3版本带扩展属性的证书 req_extensions = v3_req # 设置默认域名 [ req_distinguished_name ] # Minimum of 4 bytes are needed for common name. 注意，这里设置对应的访问域名. # 如下，指定后缀为test.com的域名都可以使用此证书访问 commonName = *.test.com commonName_default = *.test.com commonName_max = 64 # 设置两位国家代码 # ISO2 country code only countryName = China countryName_default = CN # 设置州 或者 省的名字 # State is optional, no minimum limit stateOrProvinceName = Province stateOrProvinceName_default = Guangdong # 设置城市的名字 # City is required localityName = City localityName_default = Chaoshan # 设置公司或组织机构名称 # Organization is optional organizationName = Organization organizationName_default = test # 设置部门名称 # Organization Unit is optional organizationalUnitName = Department organizationalUnitName_default = test # 设置联系邮箱 # Email is optional emailAddress = Email emailAddress_default = xxx@test.com # 拓展信息配置 [ v3_req ] basicConstraints = CA:FALSE keyUsage = nonRepudiation, digitalSignature, keyEncipherment subjectAltName = @alt_names # 要配置的域名 [alt_names] DNS.1 = *.test.com
生成好后，依次执行以下命令来生成域名jks：
#生成key openssl genrsa -out mykey.key 2048 #根据上面生成的key，生成csr文件 openssl req -new -nodes -out mycsr.csr -key mykey.key -config example.com.conf openssl req -text -noout -in mycsr.csr openssl x509 -req -days 36500 -in mycsr.csr -signkey mykey.key -out mycert.cert -extensions v3_req -extfile example.com.conf openssl x509 -noout -modulus -in mycert.cert openssl rsa -noout -modulus -in mykey.key openssl req -noout -modulus -in mycsr.csr openssl pkcs12 -export -in mycert.cert -inkey mykey.key -out presto.p12 keytool -importkeystore -srckeystore presto.p12 -srcstoretype PKCS12 -destkeystore presto.jks
在基础搭建完presto-gateway后，通过命令:
java -jar gateway-ha-1.8.8-SNAPSHOT-jar-with-dependencies.jar server ../gateway-ha-config.yml
当然你也可以编写一个脚本，直接帮你启动presto-gateway，这里我将我写的给出来：
#!/bin/bash
p=$(jps -l | grep gateway-ha | awk '{print $1}')
if [ -n "$p" ];then
echo "success kill prestogateway id : $p"
kill -9 $p
fi
#nohup java -jar -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=0.0.0.0:8780 gateway-ha-1.8.8-SNAPSHOT-jar-with-dependencies.jar server ../gateway-ha-config.yml > log.log 2>&1 &
nohup java -jar gateway-ha-1.8.8-SNAPSHOT-jar-with-dependencies.jar server ../gateway-ha-config.yml > log.log 2>&1 &
注释掉的这一行是开启debug端口的。
启动你的presto-gateway。你可以通过文档所说的rest api添加路由配置。添加完成后，这里主要讲几个presto-gateway的功能点
1、健康检查Presto-gateway每分钟会检查后端Presto集群是否健康，如果发现不正常，它会自动发送邮件通知，相关设置在配置文件 gateway-ha-config.yml中，如下：
notifier:
smtpHost: xxx.xxx.com
smtpPort: 587
sender: presto-gateway
recipients:
- xxx@xxx.com
2、优雅关闭指定的presto集群
将presto-gateway中指定的backend设置成非激活状态，这将防止任何新的传入查询被分配到后端。假设我要关闭name为presto-test的集群：
curl -X POST http://localhost:8080/gateway/backend/deactivate/presto-test
轮询Presto backend的coordinator URL，直到排队的查询计数和运行的查询计数都为0。
终止Presto Coordinator & Worker Java进程。
3、路由请求启用路由功能
如果你希望启用你自己定义的路由规则，你需要在gateway-ha-config.yml 中设置rulesEngineEnabled 为true并指明你定义的规则yml文件地址
routingRules: rulesEngineEnabled: true rulesConfigPath: "src/test/resources/rules/routing_rules.yml" # replace with path to your rules config file
缺省情况下，presto-gateway读取X-Trino-Routing-Group请求头来路由请求。如果没有指定这个报头，请求被发送到默认路由组(adhoc)。
路由规则引擎特性使您能够根据请求信息(如任何请求头)编写自定义逻辑来路由请求。路由规则与presto-gateway应用程序代码分离到配置文件中，允许动态更改规则。
定义路由规则
为了更好的触发路由规则，我们使用了easy-rules 引擎。这些规则应该存储在YAML文件中。规则由name、description、condition和actions(操作列表)组成。如果特定规则的条件计算结果为true，则触发其动作。
---
name: "cetus_adhoc_search"
description: "if query from cetus adhoc search, route to adhoc group"
condition: "request.getHeader("X-Trino-Source") == "cetus""
actions:
- "result.put("routingGroup", "adhoc")"
---
name: "airflow special"
description: "if query from airflow with special label, route to etl-special group"
condition: "request.getHeader("X-Trino-Source") == "airflow" && request.getHeader("X-Trino-Client-Tags") contains "label=special""
actions:
- "result.put("routingGroup", "etl-special")"
注意，请求头可以通过 HttpServletRequest 或者nginx转发设置。每个规则必须至少有一个action，表示前面的condition满足后，要执行的步骤。例如上面的cetus_adhoc_search，定义表明如果请求来源于cetus，就转发路由给adhoc presto集群组处理。
condition和actions编写的语法遵循mael语法。
The condition and actions are written in MVEL, an expression language with Java-like syntax. In most cases, users can write their conditions/actions in Java syntax and expect it to work. There are some MVEL-specific operators that could be useful though. For example, instead of doing a null-check before accessing the String.contains method like this:
condition: "request.getHeader("X-Trino-Client-Tags") != null && request.getHeader("X-Trino-Client-Tags").contains("label=foo")"You can use the contains operator
condition: "request.getHeader("X-Trino-Client-Tags") contains "label=foo""
注意再提醒一遍，如果没有任何规则满足，则默认转发到adhoc组！
路由优先级
如果有多个路由规则，并且请求都满足这些规则，那就按照规则的先后顺序执行。以最后一个执行的结果的actions为主去路由。
当然我们可以为规则分配一个整数值优先级。这个整数越低，它将越早触发。如果不指定优先级，则默认为INT_MAX。
---
name: "airflow"
description: "if query from airflow, route to etl group"
priority: 0
condition: "request.getHeader("X-Trino-Source") == "airflow""
actions:
- "result.put("routingGroup", "etl")"
---
name: "airflow special"
description: "if query from airflow with special label, route to etl-special group"
priority: 1
condition: "request.getHeader("X-Trino-Source") == "airflow" && request.getHeader("X-Trino-Client-Tags") contains "label=special""
actions:
- "result.put("routingGroup", "etl-special")"
只需要加上priority关键字即可。
注意，这两个规则仍然会生效。不同的是，我们保证了第一个规则(优先级0)在第二个规则(优先级1)之前被触发。因此，routingGroup被设置为etl，然后设置为etl-special，所以最终routingGroup将始终是etl-special。
Composite Rules
为了解决当一个路由请求满足多个rules时，会按照优先级次序依次执行规则actions的这种情况（出现许多无用的执行），符合规则的概念就被提出来了，方便我们更好的去控制我们想要执行的规则。
你可以参考easy-rule的复合规则文档，了解其中如何定义的：
https://github.com/j-easy/easy-rules/wiki/defining-rules#composite-rules
有了复合规则，我们可以使用ActivationRuleGroup来实现当匹配到第一个规则的时候触发执行它的actions，其余的忽略掉。
---
name: "airflow rule group"
description: "routing rules for query from airflow"
compositeRuleType: "ActivationRuleGroup"
composingRules:
- name: "airflow special"
description: "if query from airflow with special label, route to etl-special group"
priority: 0
condition: "request.getHeader("X-Trino-Source") == "airflow" && request.getHeader("X-Trino-Client-Tags") contains "label=special""
actions:
- "result.put("routingGroup", "etl-special")"
- name: "airflow"
description: "if query from airflow, route to etl group"
priority: 1
condition: "request.getHeader("X-Trino-Source") == "airflow""
actions:
- "result.put("routingGroup", "etl")"
如上，按照优先级，假设一个请求匹配了airflow rule group这个规则，那么它就会执行其中的action，并路由请求，不会走到下一个去匹配执行。
4、查询历史更多的请参考https://github.com/lyft/presto-gateway 一看就会，很easy
在添加Presto集群之后，客户端就可以通过:8080来提交查询请求，就像连接原来的Presto一样。提交查询之后，就可以在Web UI上看到历史：http://:8080/
从底部的统计可以看到查询可以平摊到两个集群上。
30分钟失效，如果失效了，就需要访问presto集群的/v1/query/{queryId}重新获取这条queryId所属的presto集群地址。
负载均衡策略在RoutingManager中的provideAdhocBackend（）方法。默认通过Random函数取模当前group的presto集群连接的个数。
三、presto多集群切换方案现在有两个presto集群，一个线上，一个预言。presto晚间业务使用量非常少，留在线上会占用yarn宝贵的资源。且晚间也会跑很多hive任务，需要资源多。为此我们考虑写脚本做线上presto集群的上线or下线。
目的：夜间将线上的presto集群做下线，切换到使用预言集群调度。白天启动线上presto集群，和预言集群一同调度任务。
定义一个shell脚本，大体的任务是：
1、每天早上7点上线启动presto线上集群。对启动的实例做检测。保证至少有1个worker启动成功（否则负载过去会报错）
具体实现：
#!/usr/bin/env python
#coding=utf-8
import os
import requests
import json
import time
import sendemail as semail
def init_variable():
global cookie
global checkStatsUri
global loginUri
global loginParam
global online_httphostAndPort
global sliderBinPath
global prestoGatewayUrl
global gatewayPrestoName
global online_checkStatsUrl
global online_loginUrl
#common config
cookie=None
checkStatsUri="/ui/api/stats"
#loginUri="/ui/login?username=presto"
loginUri="/ui/login"
loginParam={'username':'presto'}
#gateway config
gatewayPrestoName="presto-online"
prestoGatewayUrl="https://gatewayhost:8080"
#slider config
sliderBinPath="/opt/presto/slider/slider-0.92.0-incubating/bin/slider"
#presto coordinator config
online_httphostAndPort="http://${presto_online_host}:${presto_online_port}"
online_checkStatsUrl=online_httphostAndPort+checkStatsUri
online_loginUrl=online_httphostAndPort+loginUri
def printResult(res):
for line in res:
if line=='n':
continue
else:
print(line)
def krbLogin():
cmd="kinit -kt /etc/keytabs/presto.keytab presto"
#printResult(os.popen(cmd).readlines())
os.popen(cmd)
def checkWorkerStatus(tag,checkStatsUrl,loginUrl,tryTimes):
# login and get the cookie which return by server
res=requests.post(loginUrl,loginParam,allow_redirects=False)
if res != None and res.cookies != None:
cookie="Trino-UI-Token="+requests.utils.dict_from_cookiejar(res.cookies)["Trino-UI-Token"]
print(tag+":"+cookie)
if cookie is None:
raise Exception(tag+" login cookie is null")
# if activeworker <=0 ,just try more time
header={}
header["Cookie"]=cookie
res=requests.get(url=checkStatsUrl,headers=header)
workerStatus=json.loads(res.content)
activeWorkers=workerStatus["activeWorkers"]
if activeWorkers<=0:
while tryTimes>0:
tryTimes=tryTimes-1
if tryTimes>=0:
print(tag+" activeWorkers num small than 1,now is:"+str(activeWorkers))
time.sleep(10)
res=requests.get(url=checkStatsUrl,headers=header)
activeWorkers=json.loads(res.content)["activeWorkers"]
if activeWorkers<=0:
continue
else:
break
print(tag+" active workers:"+str(json.loads(res.content)))
if tryTimes<=0 and activeWorkers<=0:
raise Exception("failed to find any activeWorkers from "+checkStatsUrl)
return workerStatus
def start_online_sliderappmaster():
cmd="python "+sliderBinPath+" start trino-query"
printResult(os.popen(cmd))
#second: wait 1min
time.sleep(60)
print("wait 1 min to start workers, please be patient ....")
checkWorkerStatus(online_httphostAndPort,online_checkStatsUrl,online_loginUrl,50)
def active_gateway_online_presto():
cmd="curl -X POST "+prestoGatewayUrl+"/gateway/backend/activate/"+gatewayPrestoName+" --insecure"
printResult(os.popen(cmd))
def main():
#first: execute slider to start onlinecluster
start_online_sliderappmaster()
#second: check worker status，try 30 times until online cluster complete
workerStaus=checkWorkerStatus(online_httphostAndPort,online_checkStatsUrl,online_loginUrl,30)
print("precluster workers status:"+json.dumps(workerStaus))
#final: active gateway to distribute task to online cluster
active_gateway_online_presto()
print("execute finished")
if __name__ == '__main__':
init_variable()
krbLogin()
main()
脚本做了几件事：
- 通过slider将presto worker在线上yarn集群启动
- 探测worker的启动状态，必须确保有可用节点。
- 刷新presto_gateway对线上presto的使用状态。重置为actvie。此时请求就可以到线上集群了。
上面的error处理，你也可以发短信提醒 or 发邮件提醒，或者通过飞书or钉钉机器人推送异常，及时介入排查。
2、每天晚上12点后。先确保预言presto集群存在可用的实例数，下线presto线上集群，确保线上集群的任务跑完后再关掉。
#!/usr/bin/env python
#coding=utf-8
import os
import requests
import json
import time
def init_variable():
global cookie
global checkStatsUri
global loginUri
global loginParam
global online_httphostAndPort
global sliderBinPath
global prestoGatewayUrl
global gatewayPrestoName
global online_checkStatsUrl
global online_loginUrl
global pre_httphostAndPort
global pre_loginUrl
global pre_checkStatsUrl
#common config
cookie=None
checkStatsUri="/ui/api/stats"
loginUri="/ui/login"
loginParam={'username':'presto'}
#gateway config
gatewayPrestoName="presto-online"
prestoGatewayUrl="https://${gatewayhost}:${gatewayport}"
#slider config
sliderBinPath="/opt/presto/slider/slider-0.92.0-incubating/bin/slider"
#presto coordinator config
online_httphostAndPort="http://${presto_online_host}:${presto_online_port}"
online_checkStatsUrl=online_httphostAndPort+checkStatsUri
online_loginUrl=online_httphostAndPort+loginUri
#precluster config
pre_httphostAndPort="http://${presto_online2_host}:${presto_online2_port}"
pre_checkStatsUrl=pre_httphostAndPort+checkStatsUri
pre_loginUrl=pre_httphostAndPort+loginUri
def printResult(res):
for line in res:
if line=='n':
continue
else:
print(line)
def krbLogin():
cmd="kinit -kt /etc/keytabs/presto.keytab presto"
#printResult(os.popen(cmd).readlines())
os.popen(cmd)
def checkWorkerStatus(tag,checkStatsUrl,loginUrl,tryTimes):
# login and get the cookie which return by server
res=requests.post(loginUrl,loginParam,allow_redirects=False)
if res != None and res.cookies != None:
cookie="Trino-UI-Token="+requests.utils.dict_from_cookiejar(res.cookies)["Trino-UI-Token"]
print(tag+":"+cookie)
if cookie is None:
raise Exception(tag+" login cookie is null")
# if activeworker <=0 ,just try more time
header={}
header["Cookie"]=cookie
res=requests.get(url=checkStatsUrl,headers=header)
workerStatus=json.loads(res.content)
activeWorkers=workerStatus["activeWorkers"]
if activeWorkers<=0:
while tryTimes>0:
tryTimes=tryTimes-1
if tryTimes>=0:
print(tag+" activeWorkers num small than 1,now is:"+str(activeWorkers))
time.sleep(10)
res=requests.get(url=checkStatsUrl,headers=header)
activeWorkers=json.loads(res.content)["activeWorkers"]
if activeWorkers<=0:
continue
else:
break
print(tag+" active workers:"+str(json.loads(res.content)))
if tryTimes<=0 and activeWorkers<=0:
raise Exception("failed to find any activeWorkers from "+checkStatsUrl)
return workerStatus
def stop_online_sliderappmaster():
cmd="python "+sliderBinPath+" stop trino-query"
printResult(os.popen(cmd))
def start_online_sliderappmaster():
cmd="python "+sliderBinPath+" start trino-query"
printResult(os.popen(cmd))
checkWorkerStatus(online_httphostAndPort,online_checkStatsUrl,online_loginUrl,50)
#check and ensure the online cluster stop
def deactive_gateway_online_presto():
cmd="curl -X POST "+prestoGatewayUrl+"/gateway/backend/deactivate/"+gatewayPrestoName+" --insecure"
os.popen(cmd)
print("deactivated presto-online")
workerStatus=checkWorkerStatus(online_httphostAndPort,online_checkStatsUrl,online_loginUrl,20)
queuedQueries=int(workerStatus["queuedQueries"])
blockedQueries=int(workerStatus["blockedQueries"])
runningQueries=int(workerStatus["runningQueries"])
while queuedQueries>0 or blockedQueries>0 or runningQueries>0:
print(json.dumps(workerStatus))
time.sleep(5)
workerStatus=checkWorkerStatus(online_httphostAndPort,online_checkStatsUrl,online_loginUrl,20)
queuedQueries=int(workerStatus["queuedQueries"])
blockedQueries=int(workerStatus["blockedQueries"])
runningQueries=int(workerStatus["runningQueries"])
print("finished checked workerstatus:"+json.dumps(workerStatus)+" begin stop slider master")
stop_online_sliderappmaster()
print("success stop the online slider appmaster")
def active_gateway_online_presto():
cmd="curl -X POST "+prestoGatewayUrl+"/gateway/backend/activate/"+gatewayPrestoName+" --insecure"
printResult(os.popen(cmd))
def main():
#first: check precluster workers is still alive
workerStaus=checkWorkerStatus(pre_httphostAndPort,pre_checkStatsUrl,pre_loginUrl,5)
print("precluster workers status:"+json.dumps(workerStaus))
#second: stop the online workers
deactive_gateway_online_presto()
print("execute finished")
if __name__ == '__main__':
init_variable()
krbLogin()
main()
这个脚本主要用来做晚间下线线上presto集群的，腾出资源给其他任务使用。下线的时候会探测备用的presto集群是否有可用节点，确保下线不影响到业务正常使用。
暑期编程PK赛 得CSDN机械键盘等精美礼品！
