Hogyan hozzunk létre saját autoszkalálónkat a klaszterhez

Helló! Nagy adatokkal dolgozók képzésével foglalkozunk. Nem lehet elképzelni egy nagy adatokra vonatkozó oktatási programot a saját klaszterünk nélkül, ahol az összes résztvevő közösen dolgozik. Ennek érdekében mindig van klaszter programunk 🙂 Foglalkozunk annak beállításával, finomhangolásával és adminisztrálásával, míg a résztvevők közvetlenül futtatják a MapReduce munkákat és használják a Spark-ot.

Ebben a bejegyzésben arról beszélünk, hogyan oldottuk meg a klaszter egyenlőtlen terhelésének problémáját, saját autoszkalálót írva, felhasználva a felhőt. Mail.ru Cloud Solutions.

Probléma

Klaszterünket nem teljesen tipikus módon használjuk. A kihasználtság erősen egyenetlen. Például vannak gyakorlati órák, amikor a 30 résztvevő és az oktató belép a klaszterre, és elkezdik használni. Vagy ismét vannak napok a határidő előtt, amikor a terhelés drámaian megnő. Minden más időben a klaszter alulteherelt üzemmódban működik.

1. megoldás – egy olyan klaszter fenntartása, amely elviseli a csúcs terheléseket, de más időszakokban tétlen marad.

2. megoldás – egy kis klaszter fenntartása, amelybe kézzel adunk hozzá csomópontokat az órák előtt és a csúcs terhelések idején.

3. megoldás – egy kis klaszter fenntartása és egy autoszkaláló írása, amely figyelemmel kíséri a klaszter aktuális terhelését, és automatikusan, különböző API-kat használva, hozzáadja és eltávolítja a csomópontokat a klaszterből.

Ebben a bejegyzésben a 3. megoldásról fogunk beszélni. Ez az autoszkaláló erősen függ a külső tényezőktől, nem a belsőktől, és a szolgáltatók gyakran nem kínálják. A Mail.ru Cloud Solutions felhőinfrastruktúráját használjuk, és írtunk egy autoszkalálót az MCS API-jának felhasználásával. Mivel az adatokkal való munkára tanítunk, úgy döntöttünk, hogy bemutatjuk, hogyan írhat hasonló autoszkalálót a saját céljaidra, és használhatod a saját felhődben.

Előfeltételek

Először is szükséged van egy Hadoop klaszterre. Mi például az HDP disztribúciót használjuk.

Ahhoz, hogy a csomópontok gyorsan hozzáadhatók és eltávolíthatók legyenek, meg kell határozni bizonyos szereposztásokat a csomópontok között.

1. Mester csomópont. Itt nem igazán szükséges különösebben magyarázni: a klaszter fő csomópontja, ahol például a Spark driver indítása történik, ha interaktív módot használsz.
2. Dátaklaszter. Ez az a klaszter, ahol az adatok HDFS-en tárolódnak, és ahol a számítások is zajlanak.
3. Számítási klaszter. Ez az a klaszter, ahol az HDFS-en nincs tárolva semmi, de itt történnek a számítások.

Fontos megjegyzés. Az automatikus skálázás a harmadik típusú klaszterek révén fog megtörténni. Ha elkezdi eltávolítani és hozzáadni a második típusú klasztereket, akkor a reagálás sebessége jelentősen lelassul – a kiszervezés és újraszervezés órákig eltarthat az Ön klaszterében. Ez nyilvánvalóan nem az, amit az automatikus skálázástól vár. Tehát az első és második típusú klasztereket nem érintjük. Ezek alkotják a minimálisan életképes klasztert, amely a program időtartama alatt fennáll.

Tehát az automatikus skálázónk Python 3-ban íródott, használja az Ambari API-t a klaszterszolgáltatások kezelésére, és a Mail.ru Cloud Solutions API-t (MCS) a gépek indítására és leállítására.

A megoldás architektúrája

1. Modul autoscaler.py. Három osztályt tartalmaz: 1) funkciók az Ambari kezelésére, 2) funkciók az MCS kezelésére, 3) funkciók, amelyek közvetlenül az automatikus skálázó működési logikájához kapcsolódnak.
2. Script observer.py. Alapvetően különböző szabályokból áll: mikor és milyen pillanatokban hívják meg az automatikus skálázó funkcióit.
3. A konfigurációs paraméterek fájlja config.py. Itt található például a klaszterek listája, amelyek engedélyezettek az automatikus skálázáshoz és más paraméterek, amelyek befolyásolják, hogy mennyi időt kell várni attól a pillanattól, amikor egy új klaszter hozzáadódott. Itt találhatók a tanórák kezdetének időbélyegei is, hogy a maximálisan engedélyezett klaszterkonfiguráció elkezdését megelőzően elinduljon.

Most nézzük meg a kód darabjait, amelyek az első két fájlban találhatók.

1. autoscaler.py modul

Ambari osztály

Itt van egy kódrészlet, amely tartalmazza az Ambari osztályt.:

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": "Minden gazda komponens leállítása",
                "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 = 'Kérés elfogadva'
        else:
            message = req.status_code
        return message

A fenti példában megtekintheti a funkció megvalósítását stop_all_services, amely leállítja az összes szolgáltatást a klaszter megfelelő nodáján.

A class-nak a bemeneti paraméterei Ambari osztályt. a következők:

• ambari_url, például a következő formában: 'http://localhost:8080/api/v1/clusters/',
• cluster_name – az Ambari klaszterének neve,
• headers = {'X-Requested-By': 'ambari'}
• és itt van auth az Ambari-hoz tartozó felhasználónév és jelszó: auth = ('login', 'password').

A funkció valójában pár REST API hívás az Ambarihoz. Logikailag először megszerezzük a nodán futó szolgáltatások listáját, majd megkérjük, hogy a klaszterben, ezen a nodán változtassa meg a szolgáltatásokat "INSTALLED" állapotra. INSTALLED. Az összes szolgáltatás indítására, a nodák karbantartási állapotba helyezésére és más dolgokra vonatkozó funkciók hasonló módon épülnek fel – ezek egyszerűen több API hívás. A Mcs osztály 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_code

Itt van egy kódrészlet, amely tartalmazza az itt megadjuk a projekt azonosítóját a felhőben és a felhasználó azonosítóját, valamint a jelszavát. A funkció:

vm_turn_on

A class-nak a bemeneti paraméterei itt megadjuk a projekt azonosítóját a felhőben és a felhasználó azonosítóját, valamint a jelszavát. A funkció мы передаем id проекта внутри облака и id пользователя, а также его пароль. В функции vm_turn_on Szeretnénk bekapcsolni az egyik gépet. Itt a logika egy kicsit bonyolultabb. A kód elején három másik függvény hívása történik: 1) szükségünk van egy tokenre, 2) át kell konvertálnunk a hostname-t a MCS gépnevére, 3) meg kell szereznünk ennek a gépnek az azonosítóját. Ezután egyszerűen post-kérést küldünk, és elindítjuk ezt a gépet.

Íme a token megszerzésére szolgáló függvény:

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.token

Autoscaler osztály

Ebben az osztályban találhatóak a munkafolyamat logikájához kapcsolódó függvények.

Íme ennek az osztálynak egy részlete:

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 message

Az osztályokat, a skálázásra engedélyezett csomópontok listáját és a csomópontok konfigurációs paramétereit: a YARN-ban a csomópontra egy laptopra rendelt memória és CPU-t fogadunk. Van továbbá két belső paraméter, a q_ram és a q_cpu, amelyek sorok. Ezek segítségével tároljuk a klaszter aktuális terheltségi értékeit. Ha az utolsó 5 percben stabilan magas terhelést észlelünk, akkor úgy döntünk, hogy +1 csomópontot kell hozzáadnunk a klaszterhez. Ugyanez igaz a klaszter alulterheltségi állapotára is. Ambari osztályt. és itt megadjuk a projekt azonosítóját a felhőben és a felhasználó azonosítóját, valamint a jelszavát. A funkcióA fenti kód egy példát mutat be, amely eltávolít egy gépet a klaszterből, és leállítja azt a felhőben. Először is, leállítás történik.

YARN Nodemanager , majd bekapcsoljuk a, ezután leállítjuk az összes szolgáltatást a gépen, és kikapcsoljuk a virtuális gépet a felhőben. A Mcs osztály2. observer.py szkript

2. Скрипт observer.py

Példa kód onnan:

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} sikeresen felskálázva lett".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)

Ebben ellenőrizzük, hogy teljesültek-e a feltételek a klaszter teljesítményének növelésére és hogy van-e tartalék gépünk, megszerezzük az egyik gép nevét, hozzáadjuk a klaszterhez, és elküldünk egy üzenetet a csapat Slackjére. Ezután elindul cooldown_period, amikor semmit sem adunk hozzá, semmit sem távolítunk el a klaszterből, csak figyeljük a terhelést. Ha az stabilizálódik és az optimális terhelési értékek között van, akkor folytatjuk a megfigyelést. Ha azonban egy node nem elegendő, hozzáadunk még egyet.

Abban az esetben, ha előre tudjuk, hogy nem lesz elegendő egy node, azonnal elindítjuk az összes szabad node-ot és aktívan tartjuk őket az esemény végéig. Ez eseményidőbélyeg lista segítségével történik.

Zárás

Az autoszkálázó egy jó és kényelmes megoldás, ha a klaszter terhelése egyenetlen. Egyszerre elérheted a szükséges klaszterkonfigurációt csúcsidőszakok alatt, miközben nem tartod a klasztert alulterhelt időszakokban, így pénzt takarítasz meg. Ráadásul mindez automatizáltan történik a te részvételed nélkül. Magának az autoszkálázónak nem kell többnek lennie, mint az API hívások halmazának a klaszterkezelőhöz és a felhőszolgáltatóhoz, egy bizonyos logika szerint megszerkesztve. Amit mindenképpen meg kell jegyzni – az a node-ok 3 típusra való felosztása, ahogyan korábban írtuk. És boldog leszel.

Forrás: habr.com