Olá! Nós ensinamos as pessoas a trabalhar com grandes dados. É impossível imaginar um programa educacional de grandes dados sem seu próprio cluster, onde todos os participantes trabalham juntos. Por essa razão, ele está sempre presente em nosso programa 🙂 Nós nos encarregamos da sua configuração, ajuste e administração, enquanto os alunos executam trabalhos MapReduce e utilizam o Spark.
Neste post, contaremos como resolvemos o problema da carga desbalanceada do cluster, escrevendo nosso próprio escalador automático usando a nuvem. .
O problema
Nosso cluster não é utilizado de forma típica. A utilização é bastante desigual. Por exemplo, há aulas práticas quando todas as 30 pessoas e o professor acessam o cluster e começam a usá-lo. Ou há dias antes dos prazos, quando a carga aumenta significativamente. Em todo o resto do tempo, o cluster opera em um estado de subutilização.
Solução nº 1 – manter um cluster que suporte picos de carga, mas que fique ocioso em todo o resto do tempo.
Solução nº 2 – ter um pequeno cluster, ao qual se adicionam nós manualmente antes das aulas e durante os picos de carga.
Solução nº 3 – manter um pequeno cluster e escrever um escalador automático que monitore a carga atual do cluster e, usando várias APIs, adicione e remova nós do cluster automaticamente.
Neste post, falaremos sobre a solução nº 3. Esse escalador automático depende fortemente de fatores externos, em vez de internos, e os provedores frequentemente não o disponibilizam. Nós usamos a infraestrutura em nuvem Mail.ru Cloud Solutions e escrevemos um escalador automático usando a API MCS. E como ensinamos a trabalhar com dados, decidimos mostrar como você pode escrever um escalador automático semelhante para suas necessidades e usá-lo com sua nuvem.
Pré-requisitos
Em primeiro lugar, você deve ter um cluster Hadoop. Nós, por exemplo, utilizamos a distribuição HDP.
Para que você possa adicionar e remover nós rapidamente, é preciso ter uma distribuição de funções definida entre os nós.
- Nó mestre. Aqui não é necessário explicar muito: o nó principal do cluster, onde, por exemplo, o driver do Spark é executado, se você estiver usando o modo interativo.
- Nó de dados. Este é o nó onde seus dados são armazenados no HDFS e onde também ocorrem os cálculos.
- Nó de computação. Esta é uma nó na qual você não armazena nada no HDFS, mas onde ocorrem os cálculos.
Ponto importante. A autoescalabilidade ocorrerá com base nas nódulos do terceiro tipo. Se você começar a remover e adicionar nódulos do segundo tipo, a velocidade de resposta será muito baixa – o descomissionamento e recomissionamento levarão horas em seu cluster. Isso, claro, não é o que você espera da autoescalabilidade. Ou seja, não tocamos nas nódulos do primeiro e segundo tipo. Eles representarão um cluster minimamente viável, que existirá durante toda a execução do programa.
Assim, nosso autoescalador foi escrito em Python 3, utiliza a API do Ambari para gerenciar os serviços do cluster, usa (MCS) para iniciar e parar máquinas.
Arquitetura da Solução
- Módulo
autoscaler.py. Ele contém três classes: 1) funções para trabalhar com Ambari, 2) funções para trabalhar com MCS, 3) funções relacionadas diretamente à lógica de funcionamento do autoescalador. - Script
observer.py. Essencialmente, consiste em diferentes regras: quando e em quais momentos chamar as funções do autoescalador. - O arquivo com os parâmetros de configuração
config.py. Contém, por exemplo, uma lista de nódulos permitidos para autoescalabilidade e outros parâmetros que influenciam, por exemplo, quanto tempo esperar a partir do momento em que um novo nódulo foi adicionado. Também contém timestamps de início de atividades, para que a configuração máxima permitida do cluster esteja em execução antes da atividade.
Vamos agora dar uma olhada nos trechos de código localizados nos primeiros dois arquivos.
1. Módulo autoscaler.py
Classe Ambari
Este é um trecho de código contendo a classe Ambari:
class Ambari:
def __init__(self, ambari_url, cluster_name, headers, auth):
self.ambari_url = ambari_url
self.cluster_name = cluster_name
self.headers = headers
self.auth = auth
def stop_all_services(self, hostname):
url = self.ambari_url + self.cluster_name + '/hosts/' + hostname + '/host_components/'
url2 = self.ambari_url + self.cluster_name + '/hosts/' + hostname
req0 = requests.get(url2, headers=self.headers, auth=self.auth)
services = req0.json()['host_components']
services_list = list(map(lambda x: x['HostRoles']['component_name'], services))
data = {
"RequestInfo": {
"context":"Parar Todos os Componentes do Host",
"operation_level": {
"level":"HOST",
"cluster_name": self.cluster_name,
"host_names": hostname
},
"query":"HostRoles/component_name.in({0})".format(",".join(services_list))
},
"Body": {
"HostRoles": {
"state":"INSTALLED"
}
}
}
req = requests.put(url, data=json.dumps(data), headers=self.headers, auth=self.auth)
if req.status_code in [200, 201, 202]:
message = 'Requisição aceita'
else:
message = req.status_code
return messageAcima, como exemplo, você pode ver a implementação da função stop_all_services, que para todos os serviços no nó desejado do cluster.
Para a classe, Ambari você passa:
ambari_url, por exemplo, no formato'http://localhost:8080/api/v1/clusters/',cluster_name– o nome do seu cluster no Ambari,headers = {'X-Requested-By': 'ambari'}- e dentro
authestá o seu login e senha do Ambari:auth = ('login', 'password').
A função em si consiste em apenas algumas chamadas pela API REST do Ambari. Do ponto de vista lógico, primeiro obtemos a lista de serviços em execução no nó e depois pedimos para que esses serviços na referida cluster e nó sejam alterados para o estado INSTALLED. Funções para iniciar todos os serviços, para mudar nós para o estado Maintenance e outras, funcionam de maneira semelhante – são apenas algumas requisições pela API.
Classe Mcs
Este é um trecho de código contendo a classe Mcs:
class Mcs:
def __init__(self, id1, id2, password):
self.id1 = id1
self.id2 = id2
self.password = password
self.mcs_host = 'https://infra.mail.ru:8774/v2.1'
def vm_turn_on(self, hostname):
self.token = self.get_mcs_token()
host = self.hostname_to_vmname(hostname)
vm_id = self.get_vm_id(host)
mcs_url1 = self.mcs_host + '/servers/' + self.vm_id + '/action'
headers = {
'X-Auth-Token': '{0}'.format(self.token),
'Content-Type': 'application/json'
}
data = {'os-start' : 'null'}
mcs = requests.post(mcs_url1, data=json.dumps(data), headers=headers)
return mcs.status_codePara a classe, Mcs passamos o id do projeto dentro da nuvem e o id do usuário, assim como sua senha. Na função vm_turn_on Queremos ativar uma das máquinas. A lógica aqui é um pouco mais complexa. No início do código, há a chamada de três outras funções: 1) precisamos obter um token, 2) precisamos converter o hostname no nome da máquina no MCS, 3) obter o id dessa máquina. Em seguida, fazemos simplesmente uma requisição post e iniciamos essa máquina.
Assim é a própria função para obter o token:
def get_mcs_token(self):
url = 'https://infra.mail.ru:35357/v3/auth/tokens?nocatalog'
headers = {'Content-Type': 'application/json'}
data = {
'auth': {
'identity': {
'methods': ['password'],
'password': {
'user': {
'id': self.id1,
'password': self.password
}
}
},
'scope': {
'project': {
'id': self.id2
}
}
}
}
params = (('nocatalog', ''),)
req = requests.post(url, data=json.dumps(data), headers=headers, params=params)
self.token = req.headers['X-Subject-Token']
return self.tokenClasse Autoscaler
Nesta classe, estão contidas funções relacionadas à própria lógica de operações.
Assim é um trecho do código dessa classe:
class Autoscaler:
def __init__(self, ambari, mcs, scaling_hosts, yarn_ram_per_node, yarn_cpu_per_node):
self.scaling_hosts = scaling_hosts
self.ambari = ambari
self.mcs = mcs
self.q_ram = deque()
self.q_cpu = deque()
self.num = 0
self.yarn_ram_per_node = yarn_ram_per_node
self.yarn_cpu_per_node = yarn_cpu_per_node
def scale_down(self, hostname):
flag1 = flag2 = flag3 = flag4 = flag5 = False
if hostname in self.scaling_hosts:
while True:
time.sleep(5)
status1 = self.ambari.decommission_nodemanager(hostname)
if status1 == 'Request accepted' or status1 == 500:
flag1 = True
logging.info('Decomission request accepted: {0}'.format(flag1))
break
while True:
time.sleep(5)
status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
if status3 == 'INSTALLED':
flag3 = True
logging.info('Nodemaneger decommissioned: {0}'.format(flag3))
break
while True:
time.sleep(5)
status2 = self.ambari.maintenance_on(hostname)
if status2 == 'Request accepted' or status2 == 500:
flag2 = True
logging.info('Maintenance request accepted: {0}'.format(flag2))
break
while True:
time.sleep(5)
status4 = self.ambari.check_maintenance(hostname, 'NODEMANAGER')
if status4 == 'ON' or status4 == 'IMPLIED_FROM_HOST':
flag4 = True
self.ambari.stop_all_services(hostname)
logging.info('Maintenance is on: {0}'.format(flag4))
logging.info('Stopping services')
break
time.sleep(90)
status5 = self.mcs.vm_turn_off(hostname)
while True:
time.sleep(5)
status5 = self.mcs.get_vm_info(hostname)['server']['status']
if status5 == 'SHUTOFF':
flag5 = True
logging.info('VM is turned off: {0}'.format(flag5))
break
if flag1 and flag2 and flag3 and flag4 and flag5:
message = 'Success'
logging.info('Scale-down finished')
logging.info('Cooldown period has started. Wait for several minutes')
return messageRecebemos como entrada classes Ambari e Mcs, uma lista de nós que estão permitidos para escalonamento, bem como os parâmetros de configuração dos nós: memória e CPU alocadas no YARN. Existem também 2 parâmetros internos q_ram, q_cpu, que são filas. Com eles, armazenamos os valores da carga atual do cluster. Se notarmos que, nos últimos 5 minutos, a carga estava consistentemente alta, decidimos adicionar +1 nó ao cluster. O mesmo se aplica ao estado de subcarga do cluster.
No código acima, um exemplo de função que remove uma máquina do cluster e a desliga na nuvem é apresentado. Primeiro, ocorre a descomissionamento YARN Nodemanager, em seguida, o modo é ativado Maintenance, depois paramos todos os serviços na máquina e desligamos a máquina virtual na nuvem.
2. O script observer.py
Exemplo de código a partir daí:
if scaler.assert_up(config.scale_up_thresholds) == True:
hostname = cloud.get_vm_to_up(config.scaling_hosts)
if hostname != None:
status1 = scaler.scale_up(hostname)
if status1 == 'Success':
text = {"text": "{0} foi escalado com sucesso".format(hostname)}
post = {"text": "{0}".format(text)}
json_data = json.dumps(post)
req = requests.post(webhook, data=json_data.encode('ascii'), headers={'Content-Type': 'application/json'})
time.sleep(config.cooldown_period*60)Aqui verificamos se as condições para aumentar a capacidade do cluster foram atendidas e se há máquinas disponíveis, obtemos o hostname de uma delas, adicionamos ao cluster e publicamos uma mensagem no Slack da nossa equipe. Após isso, inicia-se cooldown_period, quando não adicionamos nem removemos nada do cluster, apenas monitoramos a carga. Se estabilizar e permanecer dentro da faixa de valores ideais, continuamos a monitorar. Se uma única nó não for suficiente, adicionamos mais uma.
Para situações em que sabemos antecipadamente que uma única nó não será suficiente, iniciamos todas as nó disponíveis e as mantemos ativas até o final da sessão. Isso ocorre por meio de uma lista de timestamps das sessões.
Conclusão
O autoescalador é uma solução boa e conveniente para quando você tem uma carga desigual no cluster. Você consegue a configuração necessária do cluster para cargas de pico e, ao mesmo tempo, não mantém esse cluster durante períodos de baixa carga, economizando recursos. Além disso, tudo isso acontece de forma automatizada sem a sua participação. O autoescalador em si é apenas um conjunto de chamadas à API do gerenciador de cluster e à API do provedor de nuvem, organizadas de acordo com uma lógica específica. O que você definitivamente deve lembrar é sobre a divisão das nó em 3 tipos, como mencionamos anteriormente. E isso trará felicidade para você.
Fonte: habr.com
