🥇Jinsi ya kutengeneza kihesabu otomatiki chako kwa kundi

Habari! Tunawafundisha watu kufanya kazi na data kubwa. Haiwezekani kufikiria mpango wa elimu juu ya data kubwa bila nguzo yake mwenyewe, ambayo washiriki wote hufanya kazi pamoja. Kwa sababu hii, programu yetu huwa nayo kila wakati :) Tunajishughulisha na usanidi, urekebishaji na usimamizi wake, na wavulana huzindua moja kwa moja kazi za MapReduce huko na kutumia Spark.

Katika chapisho hili tutakuambia jinsi tulivyosuluhisha shida ya upakiaji wa nguzo zisizo sawa kwa kuandika autoscaler yetu wenyewe kwa kutumia wingu. Mail.ru Cloud Solutions.

tatizo

Nguzo yetu haitumiki katika hali ya kawaida. Utupaji hauna usawa sana. Kwa mfano, kuna madarasa ya vitendo, wakati watu wote 30 na mwalimu wanakwenda kwenye nguzo na kuanza kuitumia. Au tena, kuna siku kabla ya tarehe ya mwisho wakati mzigo unaongezeka sana. Wakati uliobaki, nguzo hufanya kazi katika hali ya upakiaji.

Suluhisho # 1 ni kuweka nguzo ambayo itastahimili mizigo ya juu, lakini itakuwa bila kufanya kitu wakati wote uliobaki.

Suluhisho # 2 ni kuweka nguzo ndogo, ambayo wewe huongeza nodi kabla ya madarasa na wakati wa mizigo ya kilele.

Suluhisho # 3 ni kuweka nguzo ndogo na kuandika autoscaler ambayo itafuatilia mzigo wa sasa wa nguzo na, kwa kutumia API mbalimbali, kuongeza na kuondoa nodes kutoka kwa nguzo.

Katika chapisho hili tutazungumza juu ya suluhisho # 3. Kidhibiti kiotomatiki hiki kinategemea sana mambo ya nje badala ya yale ya ndani, na watoa huduma mara nyingi hawatoi. Tunatumia miundombinu ya wingu ya Mail.ru Cloud Solutions na tukaandika kihesabu otomatiki kwa kutumia API ya MCS. Na kwa kuwa tunafundisha jinsi ya kufanya kazi na data, tuliamua kuonyesha jinsi unavyoweza kuandika kihesabu otomatiki kwa madhumuni yako mwenyewe na kukitumia na wingu lako.

Prerequisites

Kwanza, lazima uwe na nguzo ya Hadoop. Kwa mfano, tunatumia usambazaji wa HDP.

Ili nodes zako ziongezwe haraka na kuondolewa, lazima uwe na usambazaji fulani wa majukumu kati ya nodes.

Nodi kuu. Kweli, hakuna kitu kinachohitajika kuelezea hapa: node kuu ya nguzo, ambayo, kwa mfano, dereva wa Spark huzinduliwa, ikiwa unatumia hali ya maingiliano.
Nodi ya tarehe. Hii ndio nodi ambayo unahifadhi data kwenye HDFS na ambapo mahesabu hufanyika.
Nodi ya kompyuta. Hii ni nodi ambapo hauhifadhi chochote kwenye HDFS, lakini ambapo mahesabu hufanyika.

Jambo muhimu. Autoscaling itatokea kwa sababu ya nodi za aina ya tatu. Ukianza kuchukua na kuongeza nodi za aina ya pili, kasi ya majibu itakuwa ya chini sana - kusitisha na kuwasilisha tena itachukua masaa kwenye nguzo yako. Hii, kwa kweli, sio kile unachotarajia kutoka kwa kuongeza otomatiki. Hiyo ni, hatugusa nodes za aina ya kwanza na ya pili. Zitawakilisha nguzo ya chini kabisa ambayo itakuwepo katika muda wote wa programu.

Kwa hivyo, autoscaler yetu imeandikwa katika Python 3, hutumia API ya Ambari kudhibiti huduma za nguzo, matumizi. API kutoka kwa Mail.ru Cloud Solutions (MCS) ya kuanzisha na kusimamisha mashine.

Usanifu wa suluhisho

Moduli autoscaler.py. Ina madarasa matatu: 1) kazi za kufanya kazi na Ambari, 2) kazi za kufanya kazi na MCS, 3) kazi zinazohusiana moja kwa moja na mantiki ya autoscaler.
Hati observer.py. Kimsingi ina sheria tofauti: lini na wakati gani wa kuita kazi za autoscaler.
Faili ya usanidi config.py. Ina, kwa mfano, orodha ya nodes zinazoruhusiwa kwa autoscaling na vigezo vingine vinavyoathiri, kwa mfano, muda gani wa kusubiri kutoka wakati node mpya iliongezwa. Pia kuna alama za nyakati za kuanza kwa madarasa, ili kabla ya darasa usanidi wa juu unaoruhusiwa wa nguzo uzinduliwe.

Hebu sasa tuangalie vipande vya msimbo ndani ya faili mbili za kwanza.

1. Moduli ya Autoscaler.py

Darasa la Ambari

Hivi ndivyo kipande cha nambari iliyo na darasa inavyoonekana 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 = 'Request accepted'
        else:
            message = req.status_code
        return message

Hapo juu, kama mfano, unaweza kuangalia utekelezaji wa kazi stop_all_services, ambayo inasimamisha huduma zote kwenye nodi ya nguzo inayotaka.

Katika mlango wa darasa Ambari unapita:

ambari_url, kwa mfano, kama 'http://localhost:8080/api/v1/clusters/',
cluster_name - jina la nguzo yako huko Ambari,
headers = {'X-Requested-By': 'ambari'}
na ndani auth hapa kuna kuingia kwako na nenosiri la Ambari: auth = ('login', 'password').

Kazi yenyewe sio zaidi ya simu kadhaa kupitia REST API hadi Ambari. Kutoka kwa mtazamo wa kimantiki, tunapokea kwanza orodha ya huduma zinazoendesha kwenye nodi, na kisha tunauliza kwenye nguzo fulani, kwenye nodi fulani, kuhamisha huduma kutoka kwenye orodha hadi kwa serikali. INSTALLED. Kazi za kuzindua huduma zote, kwa kuhamisha nodi kwa hali Maintenance n.k. zinafanana - ni maombi machache tu kupitia API.

Madarasa Mcs

Hivi ndivyo kipande cha nambari iliyo na darasa inavyoonekana 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

Katika mlango wa darasa Mcs tunapitisha kitambulisho cha mradi ndani ya wingu na kitambulisho cha mtumiaji, pamoja na nenosiri lake. Katika utendaji vm_turn_on tunataka kuwasha moja ya mashine. Mantiki hapa ni ngumu zaidi. Mwanzoni mwa nambari, kazi zingine tatu zinaitwa: 1) tunahitaji kupata ishara, 2) tunahitaji kubadilisha jina la mwenyeji kuwa jina la mashine katika MCS, 3) pata kitambulisho cha mashine hii. Ifuatayo, tunatuma ombi la chapisho na kuzindua mashine hii.

Hivi ndivyo kazi ya kupata ishara inaonekana kama:

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

Darasa la Autoscaler

Darasa hili lina vipengele vinavyohusiana na mantiki ya uendeshaji yenyewe.

Hivi ndivyo kipande cha nambari ya darasa hili inaonekana kama:

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

Tunakubali madarasa kwa ajili ya kuingia. Ambari и Mcs, orodha ya nodes zinazoruhusiwa kwa kuongeza, pamoja na vigezo vya usanidi wa nodi: kumbukumbu na cpu iliyotengwa kwa node katika YARN. Pia kuna vigezo 2 vya ndani q_ram, q_cpu, ambavyo ni foleni. Kwa kuzitumia, tunahifadhi maadili ya mzigo wa sasa wa nguzo. Ikiwa tunaona kwamba zaidi ya dakika 5 zilizopita kumekuwa na mzigo ulioongezeka mara kwa mara, basi tunaamua kwamba tunahitaji kuongeza nodi ya +1 kwenye nguzo. Vile vile ni kweli kwa hali ya matumizi duni ya nguzo.

Nambari iliyo hapo juu ni mfano wa kazi ambayo huondoa mashine kutoka kwa nguzo na kuisimamisha kwenye wingu. Kwanza kuna kufuta YARN Nodemanager, kisha modi inawashwa Maintenance, basi tunasimamisha huduma zote kwenye mashine na kuzima mashine ya kawaida kwenye wingu.

2. Mwangalizi wa hati.py

Mfano wa nambari kutoka hapo:

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} has been successfully scaled-up".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)

Ndani yake, tunaangalia ikiwa hali zimeundwa kwa ajili ya kuongeza uwezo wa nguzo na kama kuna mashine zozote zilizohifadhiwa, pata jina la mpangishi wa mojawapo, tuiongeze kwenye kundi na kuchapisha ujumbe kuihusu kwenye Slack ya timu yetu. Baada ya hapo huanza cooldown_period, wakati hatuongezi au kuondoa chochote kutoka kwa nguzo, lakini tu kufuatilia mzigo. Ikiwa imetulia na iko ndani ya ukanda wa maadili bora ya mzigo, basi tunaendelea kufuatilia. Ikiwa node moja haitoshi, basi tunaongeza nyingine.

Kwa kesi wakati tuna somo mbele, tayari tunajua kwa hakika kwamba node moja haitoshi, kwa hiyo sisi mara moja tunaanza nodes zote za bure na kuziweka kazi hadi mwisho wa somo. Hii hutokea kwa kutumia orodha ya alama za nyakati za shughuli.

Hitimisho

Autoscaler ni suluhisho nzuri na rahisi kwa kesi hizo wakati utapata upakiaji wa nguzo zisizo sawa. Wakati huo huo unafanikisha usanidi unaohitajika wa nguzo kwa mizigo ya kilele na wakati huo huo usihifadhi nguzo hii wakati wa upakiaji, kuokoa pesa. Kweli, pamoja na haya yote hufanyika kiotomatiki bila ushiriki wako. Autoscaler yenyewe sio zaidi ya seti ya maombi kwa API ya meneja wa nguzo na API ya mtoaji wa wingu, iliyoandikwa kulingana na mantiki fulani. Unachohitaji kukumbuka ni mgawanyiko wa nodi katika aina 3, kama tulivyoandika hapo awali. Na utakuwa na furaha.

Chanzo: mapenzi.com

Jinsi ya kutengeneza autoscaler yako mwenyewe kwa nguzo