Ciao! Insegniamo alle persone a lavorare con i big data. È impossibile immaginare un programma educativo sui big data senza un proprio cluster, su cui tutti i partecipanti lavorano insieme. Per questo motivo, nel nostro programma, è sempre presente 🙂 Ci occupiamo della sua configurazione, ottimizzazione e amministrazione, mentre i ragazzi eseguono MapReduce jobs e utilizzano Spark.
In questo post, parleremo di come abbiamo affrontato il problema del caricamento non uniforme del cluster, scrivendo il nostro autoscaler utilizzando il cloud. .
Problema
Il nostro cluster non viene utilizzato in modo tipico. L'utilizzo è molto irregolare. Ad esempio, ci sono lezioni pratiche in cui tutti e 30 gli studenti e l'insegnante accedono al cluster e iniziano a usarlo. Oppure ci sono giorni prima della scadenza in cui il carico aumenta notevolmente. In tutti gli altri momenti, il cluster funziona in modalità sottoutilizzata.
Soluzione n. 1: mantenere un cluster in grado di supportare i picchi di carico, ma che rimarrà inattivo in tutti gli altri momenti.
Soluzione n. 2: mantenere un piccolo cluster, aggiungendo manualmente nodi prima delle lezioni e durante i picchi di carico.
Soluzione n. 3: mantenere un piccolo cluster e scrivere un autoscaler che monitori il carico attuale del cluster e aggiunga e rimuova nodi dal cluster utilizzando diverse API.
In questo post parleremo della soluzione n. 3. Questo autoscaler dipende fortemente da fattori esterni, piuttosto che interni, e i fornitori non lo offrono spesso. Utilizziamo l'infrastruttura cloud di Mail.ru Cloud Solutions e abbiamo scritto un autoscaler utilizzando l'API MCS. Poiché insegniamo a lavorare con i dati, abbiamo deciso di mostrare come puoi scrivere un autoscaler simile per i tuoi scopi e usarlo con il tuo cloud.
Prerequisites
Prima di tutto, devi avere un cluster Hadoop. Noi, ad esempio, utilizziamo la distribuzione HDP.
Affinché i nodi possano essere aggiunti e rimossi rapidamente, devi avere una certa distribuzione dei ruoli sui nodi.
- Nodo master. Qui non c'è molto da spiegare: è il nodo principale del cluster, sul quale viene avviato, ad esempio, il driver di Spark, se utilizzi la modalità interattiva.
- Nodo dati. Questo è il nodo su cui vengono memorizzati i dati su HDFS e dove avvengono anche i calcoli.
- Nodo di calcolo. Questo è il nodo su cui non vengono memorizzati dati su HDFS, ma dove avvengono i calcoli.
Un punto importante. L'autoscaling avverrà tramite nodi di terzo tipo. Se inizi a rimuovere e aggiungere nodi di secondo tipo, i tempi di risposta saranno molto lenti: il decommissioning e il recommissioning richiederanno ore per il tuo cluster. Questo, naturalmente, non è ciò che ci si aspetta da un autoscaling. Pertanto, non tocchiamo i nodi di primo e secondo tipo. Essi rappresenteranno un cluster minimamente vitale che esisterà durante l'intera durata del programma.
Il nostro autoscaler è scritto in Python 3, utilizza l'API di Ambari per gestire i servizi del cluster, e utilizza (MCS) per avviare e fermare le macchine.
Architettura della soluzione
- Modulo
autoscaler.py. Esso contiene tre classi: 1) funzioni per lavorare con Ambari, 2) funzioni per lavorare con MCS, 3) funzioni legate direttamente alla logica dell'autoscaler. - Script
observer.py. Consiste essenzialmente in diverse regole: quando e in quali momenti chiamare le funzioni dell'autoscaler. - Il file con i parametri di configurazione
config.py. Esso contiene, ad esempio, un elenco di nodi consentiti per l'autoscaling e altri parametri che influenzano, ad esempio, quanto tempo aspettare dall'aggiunta di un nuovo nodo. Qui si trovano anche i timestamp di inizio delle lezioni, per garantire che prima della lezione sia stata avviata la configurazione massima consentita del cluster.
Ora diamo un'occhiata ai pezzi di codice presenti nei primi due file.
1. Modulo autoscaler.py
Classe Ambari
Ecco un pezzo di codice che contiene la 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":"Stop All Host Components",
"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 = 'Richiesta accettata'
else:
message = req.status_code
return messageIn alto, come esempio, puoi vedere l'implementazione della funzione stop_all_services, che ferma tutti i servizi sul nodo desiderato del cluster.
In input alla classe Ambari hai bisogno di fornire:
ambari_url, ad esempio, del tipo'http://localhost:8080/api/v1/clusters/',cluster_name– il nome del tuo cluster in Ambari,headers = {'X-Requested-By': 'ambari'}- e all'interno
authci sono le tue credenziali di accesso ad Ambari:auth = ('login', 'password').
La funzione stessa non è altro che un paio di richieste tramite REST API a Ambari. Dal punto di vista logico, iniziamo ricevendo un elenco dei servizi attivi sul nodo e poi chiediamo di riportare i servizi dell'elenco allo stato INSTALLED. Funzioni per avviare tutti i servizi, per mettere i nodi in stato Maintenance e altro funzionano in modo simile: sono semplicemente un insieme di richieste tramite API.
Classe Mcs
Ecco un pezzo di codice che contiene la 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_codeIn input alla classe Mcs forniamo l'id del progetto all'interno del cloud e l'id dell'utente, oltre alla sua password. Nella funzione vm_turn_on vogliamo attivare una delle macchine. La logica qui è un po' più complessa. All'inizio del codice chiamiamo tre altre funzioni: 1) abbiamo bisogno di ottenere un token, 2) dobbiamo convertire il nome hostname nel nome della macchina in MCS, 3) ottenere l'id di quella macchina. Successivamente, facciamo semplicemente una richiesta post per avviare la macchina.
Ecco come si presenta la funzione per ottenere il 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
In questa classe ci sono funzioni riguardanti la logica operativa.
Ecco come si presenta un pezzo di codice di questa 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('Richiesta di dismissione accettata: {0}'.format(flag1))
break
while True:
time.sleep(5)
status3 = self.ambari.check_service(hostname, 'NODEMANAGER')
if status3 == 'INSTALLED':
flag3 = True
logging.info('Nodemaneger dismesso: {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('Richiesta di manutenzione accettata: {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('La manutenzione è attiva: {0}'.format(flag4))
logging.info('Fermando i servizi')
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 è spenta: {0}'.format(flag5))
break
if flag1 and flag2 and flag3 and flag4 and flag5:
message = 'Successo'
logging.info('Scale-down completata')
logging.info('Periodo di raffreddamento iniziato. Attendere alcuni minuti')
return messageIn input accettiamo classi Ambari e Mcs, lista dei nodi consentiti per lo scaling e le configurazioni dei nodi: memoria e CPU allocate al nodo in YARN. Ci sono anche 2 parametri interni q_ram, q_cpu, che fungono da code. Attraverso di essi teniamo traccia dei valori di carico attuale del cluster. Se vediamo che negli ultimi 5 minuti c'è stata una carico elevato costante, prendiamo la decisione di aggiungere +1 nodo al cluster. Lo stesso vale per la condizione di sottoutilizzo del cluster.
Il codice sopra mostra un esempio di funzione che rimuove una macchina dal cluster e la ferma nel cloud. Inizialmente avviene il decommissioning YARN Nodemanager, poi si attiva la modalità Maintenance, quindi fermiamo tutti i servizi sulla macchina e spegniamo la macchina virtuale nel cloud.
2. Script observer.py
Ecco un esempio di codice da lì:
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} è stato scalato con successo".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)In esso controlliamo se le condizioni per aumentare le risorse del cluster sono soddisfatte e se ci sono macchine disponibili, otteniamo il hostname di una di esse, la aggiungiamo al cluster e pubblichiamo un messaggio a riguardo su Slack per il nostro team. Dopo di che inizia cooldown_period, durante il quale non aggiungiamo né rimuoviamo nulla dal cluster, ma monitoriamo semplicemente il carico. Se si stabilizza e rimane all'interno dell'intervallo ottimale, continuiamo a monitorare. Se una sola nodo non è sufficiente, ne aggiungiamo un'altra.
In caso di eventi programmati, sappiamo già per certo che una sola nodo non sarà sufficiente, quindi avviamo immediatamente tutte le nodi disponibili e le manteniamo attive fino alla fine dell'evento. Questo avviene tramite un elenco di timestamp degli eventi.
Conclusione
L'autoscaler è una soluzione utile e pratica per quelli casi in cui si osserva un carico irregolare nel cluster. Si ottiene la configurazione desiderata del cluster per i carichi di picco senza mantenere il cluster durante i periodi di sottoutilizzo, risparmiando così fondi. Inoltre, tutto ciò avviene in modo automatizzato senza il vostro intervento. L'autoscaler non è altro che una serie di richieste all'API del cluster manager e all'API del provider cloud, definite secondo una logica specifica. È importante ricordare la divisione dei nodi in 3 tipi, come menzionato in precedenza. E sarete felici.
Fonte: habr.com
