ક્લસ્ટર માટે તમારું પોતાનું ઓટોસ્કેલર કેવી રીતે બનાવવું

નમસ્તે! અમે લોકોને મોટા ડેટા સાથે કામ કરવાની તાલીમ આપીએ છીએ. તેના પોતાના ક્લસ્ટર વિના મોટા ડેટા પર શૈક્ષણિક પ્રોગ્રામની કલ્પના કરવી અશક્ય છે, જેના પર બધા સહભાગીઓ સાથે મળીને કામ કરે છે. આ કારણોસર, અમારા પ્રોગ્રામમાં હંમેશા તે હોય છે 🙂 અમે તેના રૂપરેખાંકન, ટ્યુનિંગ અને વહીવટમાં રોકાયેલા છીએ, અને લોકો સીધા જ ત્યાં MapReduce નોકરીઓ શરૂ કરે છે અને સ્પાર્કનો ઉપયોગ કરે છે.

આ પોસ્ટમાં અમે તમને જણાવીશું કે અમે ક્લાઉડનો ઉપયોગ કરીને અમારું પોતાનું ઓટોસ્કેલર લખીને અસમાન ક્લસ્ટર લોડિંગની સમસ્યાને કેવી રીતે હલ કરી. Mail.ru ક્લાઉડ સોલ્યુશન્સ.

સમસ્યા

અમારા ક્લસ્ટરનો ઉપયોગ લાક્ષણિક મોડમાં થતો નથી. નિકાલ અત્યંત અસમાન છે. ઉદાહરણ તરીકે, ત્યાં પ્રાયોગિક વર્ગો છે, જ્યારે બધા 30 લોકો અને એક શિક્ષક ક્લસ્ટરમાં જાય છે અને તેનો ઉપયોગ કરવાનું શરૂ કરે છે. અથવા ફરીથી, સમયમર્યાદા પહેલાના દિવસો હોય છે જ્યારે લોડ મોટા પ્રમાણમાં વધે છે. બાકીનો સમય ક્લસ્ટર અન્ડરલોડ મોડમાં કાર્ય કરે છે.

સોલ્યુશન #1 એ એક ક્લસ્ટર રાખવાનું છે જે પીક લોડનો સામનો કરશે, પરંતુ બાકીના સમયે નિષ્ક્રિય રહેશે.

સોલ્યુશન #2 એ એક નાનું ક્લસ્ટર રાખવું છે, જેમાં તમે વર્ગો પહેલાં અને પીક લોડ દરમિયાન મેન્યુઅલી નોડ્સ ઉમેરો છો.

સોલ્યુશન #3 એ એક નાનું ક્લસ્ટર રાખવું અને ઓટોસ્કેલર લખવાનું છે જે ક્લસ્ટરના વર્તમાન લોડને મોનિટર કરશે અને, વિવિધ API નો ઉપયોગ કરીને, ક્લસ્ટરમાંથી નોડ્સ ઉમેરો અને દૂર કરશે.

આ પોસ્ટમાં આપણે ઉકેલ નંબર 3 વિશે વાત કરીશું. આ ઓટોસ્કેલર આંતરિક પરિબળોને બદલે બાહ્ય પરિબળો પર ખૂબ જ નિર્ભર છે, અને પ્રદાતાઓ ઘણીવાર તે પ્રદાન કરતા નથી. અમે Mail.ru ક્લાઉડ સોલ્યુશન્સ ક્લાઉડ ઈન્ફ્રાસ્ટ્રક્ચરનો ઉપયોગ કરીએ છીએ અને MCS API નો ઉપયોગ કરીને ઑટોસ્કેલર લખીએ છીએ. અને અમે ડેટા સાથે કેવી રીતે કામ કરવું તે શીખવતા હોવાથી, અમે તે બતાવવાનું નક્કી કર્યું કે તમે તમારા પોતાના હેતુઓ માટે સમાન ઓટોસ્કેલર કેવી રીતે લખી શકો છો અને તેનો ઉપયોગ તમારા ક્લાઉડ સાથે કરી શકો છો.

પૂર્વજરૂરીયાતો

પ્રથમ, તમારી પાસે Hadoop ક્લસ્ટર હોવું આવશ્યક છે. ઉદાહરણ તરીકે, અમે HDP વિતરણનો ઉપયોગ કરીએ છીએ.

તમારા નોડ્સ ઝડપથી ઉમેરવા અને દૂર કરવા માટે, તમારી પાસે નોડ્સ વચ્ચે ભૂમિકાઓનું ચોક્કસ વિતરણ હોવું આવશ્યક છે.

1. માસ્ટર નોડ. ઠીક છે, ખાસ કરીને કંઈપણ સમજાવવાની જરૂર નથી: ક્લસ્ટરનો મુખ્ય નોડ, જેના પર, ઉદાહરણ તરીકે, જો તમે ઇન્ટરેક્ટિવ મોડનો ઉપયોગ કરો છો, તો સ્પાર્ક ડ્રાઇવર શરૂ થાય છે.
2. તારીખ નોડ. આ એ નોડ છે કે જેના પર તમે HDFS પર ડેટા સ્ટોર કરો છો અને જ્યાં ગણતરીઓ થાય છે.
3. કમ્પ્યુટિંગ નોડ. આ એક નોડ છે જ્યાં તમે HDFS પર કંઈપણ સંગ્રહિત કરતા નથી, પરંતુ જ્યાં ગણતરીઓ થાય છે.

મહત્વનો મુદ્દો. ત્રીજા પ્રકારના નોડ્સને કારણે ઑટોસ્કેલિંગ થશે. જો તમે બીજા પ્રકારના નોડ્સ લેવાનું અને ઉમેરવાનું શરૂ કરો છો, તો પ્રતિસાદની ઝડપ ઘણી ઓછી હશે - ડિકમિશનિંગ અને રિકમિટિંગ તમારા ક્લસ્ટર પર કલાકો લેશે. આ, અલબત્ત, તમે ઓટોસ્કેલિંગથી અપેક્ષા રાખતા નથી. એટલે કે, અમે પ્રથમ અને બીજા પ્રકારનાં ગાંઠોને સ્પર્શતા નથી. તેઓ ન્યૂનતમ વ્યવહારુ ક્લસ્ટરનું પ્રતિનિધિત્વ કરશે જે પ્રોગ્રામના સમગ્ર સમયગાળા દરમિયાન અસ્તિત્વમાં રહેશે.

તેથી, અમારું ઓટોસ્કેલર Python 3 માં લખાયેલું છે, ક્લસ્ટર સેવાઓનું સંચાલન કરવા માટે Ambari API નો ઉપયોગ કરે છે, Mail.ru ક્લાઉડ સોલ્યુશન્સમાંથી API (MCS) મશીનો શરૂ કરવા અને બંધ કરવા માટે.

સોલ્યુશન આર્કિટેક્ચર

1. મોડ્યુલ autoscaler.py. તેમાં ત્રણ વર્ગો છે: 1) અંબરી સાથે કામ કરવા માટેના કાર્યો, 2) MCS સાથે કામ કરવા માટેના કાર્યો, 3) ઑટોસ્કેલરના તર્ક સાથે સીધા સંબંધિત કાર્યો.
2. સ્ક્રિપ્ટ observer.py. અનિવાર્યપણે તે વિવિધ નિયમો ધરાવે છે: ક્યારે અને કઈ ક્ષણે ઑટોસ્કેલર ફંક્શનને કૉલ કરવો.
3. રૂપરેખાંકન ફાઇલ config.py. તેમાં, ઉદાહરણ તરીકે, ઓટોસ્કેલિંગ માટે માન્ય નોડ્સની સૂચિ અને અન્ય પરિમાણો છે જે અસર કરે છે, ઉદાહરણ તરીકે, નવો નોડ ઉમેરવાની ક્ષણથી કેટલો સમય રાહ જોવી. વર્ગો શરૂ કરવા માટે ટાઇમસ્ટેમ્પ્સ પણ છે, જેથી વર્ગ પહેલાં મહત્તમ અનુમતિ પ્રાપ્ત ક્લસ્ટર ગોઠવણી શરૂ કરવામાં આવે.

ચાલો હવે પ્રથમ બે ફાઈલોની અંદરના કોડના ટુકડાઓ જોઈએ.

1. Autoscaler.py મોડ્યુલ

અંબરી વર્ગ

વર્ગ સમાવિષ્ટ કોડનો ભાગ આ જેવો દેખાય છે 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

ઉપર, ઉદાહરણ તરીકે, તમે કાર્યના અમલીકરણને જોઈ શકો છો stop_all_services, જે ઇચ્છિત ક્લસ્ટર નોડ પર તમામ સેવાઓને અટકાવે છે.

વર્ગના પ્રવેશદ્વાર પર Ambari તમે પાસ કરો:

• ambari_url, ઉદાહરણ તરીકે, જેમ 'http://localhost:8080/api/v1/clusters/',
• cluster_name - અંબરીમાં તમારા ક્લસ્ટરનું નામ,
• headers = {'X-Requested-By': 'ambari'}
• અને અંદર auth અંબરી માટે તમારું લોગિન અને પાસવર્ડ આ રહ્યો: auth = ('login', 'password').

ફંક્શન પોતે REST API દ્વારા અંબરી પરના બે કોલ્સ સિવાય બીજું કંઈ નથી. તાર્કિક દૃષ્ટિકોણથી, અમે પ્રથમ નોડ પર ચાલી રહેલી સેવાઓની સૂચિ પ્રાપ્ત કરીએ છીએ, અને પછી આપેલ ક્લસ્ટર પર, આપેલ નોડ પર, સેવાઓને સૂચિમાંથી રાજ્યમાં સ્થાનાંતરિત કરવા માટે કહીએ છીએ. INSTALLED. તમામ સેવાઓ શરૂ કરવા, રાજ્યમાં નોડ્સ ટ્રાન્સફર કરવા માટેના કાર્યો Maintenance વગેરે સમાન દેખાય છે - તે API દ્વારા માત્ર થોડી વિનંતીઓ છે.

વર્ગ Mcs

વર્ગ સમાવિષ્ટ કોડનો ભાગ આ જેવો દેખાય છે 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

વર્ગના પ્રવેશદ્વાર પર Mcs અમે ક્લાઉડની અંદર પ્રોજેક્ટ આઈડી અને યુઝર આઈડી તેમજ તેનો પાસવર્ડ પાસ કરીએ છીએ. કાર્યમાં vm_turn_on અમે એક મશીન ચાલુ કરવા માંગીએ છીએ. અહીંનો તર્ક થોડો વધુ જટિલ છે. કોડની શરૂઆતમાં, અન્ય ત્રણ ફંક્શન કહેવામાં આવે છે: 1) આપણે ટોકન મેળવવાની જરૂર છે, 2) આપણે હોસ્ટનેમને MCS માં મશીનના નામમાં કન્વર્ટ કરવાની જરૂર છે, 3) આ મશીનની આઈડી મેળવો. આગળ, અમે ખાલી પોસ્ટ વિનંતી કરીએ છીએ અને આ મશીનને લોન્ચ કરીએ છીએ.

ટોકન મેળવવાનું કાર્ય આના જેવું દેખાય છે:

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

ઓટોસ્કેલર વર્ગ

આ વર્ગમાં ઓપરેટિંગ તર્ક સાથે સંબંધિત કાર્યો છે.

આ વર્ગ માટે કોડનો ભાગ આના જેવો દેખાય છે:

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

અમે પ્રવેશ માટે વર્ગો સ્વીકારીએ છીએ. Ambari и Mcs, નોડ્સની સૂચિ કે જે સ્કેલિંગ માટે માન્ય છે, તેમજ નોડ રૂપરેખાંકન પરિમાણો: મેમરી અને સીપીયુ યાર્નમાં નોડને ફાળવવામાં આવે છે. ત્યાં પણ 2 આંતરિક પરિમાણો q_ram, q_cpu છે, જે કતાર છે. તેનો ઉપયોગ કરીને, અમે વર્તમાન ક્લસ્ટર લોડના મૂલ્યોને સંગ્રહિત કરીએ છીએ. જો આપણે જોઈએ કે છેલ્લી 5 મિનિટમાં લોડમાં સતત વધારો થયો છે, તો અમે નક્કી કરીએ છીએ કે આપણે ક્લસ્ટરમાં +1 નોડ ઉમેરવાની જરૂર છે. ક્લસ્ટર અંડરયુટિલાઈઝેશન સ્ટેટ માટે પણ આવું જ છે.

ઉપરનો કોડ એ ફંક્શનનું ઉદાહરણ છે જે ક્લસ્ટરમાંથી મશીનને દૂર કરે છે અને તેને ક્લાઉડમાં રોકે છે. પ્રથમ ત્યાં ડિકમિશનિંગ છે YARN Nodemanager, પછી મોડ ચાલુ થાય છે Maintenance, પછી અમે મશીન પરની બધી સેવાઓ બંધ કરીએ છીએ અને ક્લાઉડમાં વર્ચ્યુઅલ મશીન બંધ કરીએ છીએ.

2. સ્ક્રિપ્ટ નિરીક્ષક.py

ત્યાંથી નમૂના કોડ:

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)

તેમાં, અમે તપાસ કરીએ છીએ કે ક્લસ્ટરની ક્ષમતા વધારવા માટે પરિસ્થિતિઓ વિકસિત થઈ છે કે કેમ અને અનામતમાં કોઈ મશીન છે કે કેમ, તેમાંથી એકનું હોસ્ટનામ મેળવીએ છીએ, તેને ક્લસ્ટરમાં ઉમેરીએ છીએ અને અમારી ટીમના સ્લેક પર તેના વિશે સંદેશ પ્રકાશિત કરીએ છીએ. જે પછી તે શરૂ થાય છે cooldown_period, જ્યારે અમે ક્લસ્ટરમાંથી કંઈપણ ઉમેરતા નથી અથવા દૂર કરતા નથી, પરંતુ ફક્ત લોડનું નિરીક્ષણ કરીએ છીએ. જો તે સ્થિર થઈ ગયું છે અને શ્રેષ્ઠ લોડ મૂલ્યોના કોરિડોરમાં છે, તો અમે ફક્ત દેખરેખ ચાલુ રાખીએ છીએ. જો એક નોડ પૂરતો ન હતો, તો અમે બીજો ઉમેરો.

એવા કિસ્સાઓ માટે જ્યારે અમારી પાસે પાઠ આગળ હોય, અમે પહેલાથી જ ખાતરીપૂર્વક જાણીએ છીએ કે એક નોડ પૂરતો નથી, તેથી અમે તરત જ તમામ મફત ગાંઠો શરૂ કરીએ છીએ અને પાઠના અંત સુધી તેમને સક્રિય રાખીએ છીએ. આ પ્રવૃત્તિ ટાઇમસ્ટેમ્પ્સની સૂચિનો ઉપયોગ કરીને થાય છે.

નિષ્કર્ષ

જ્યારે તમે અસમાન ક્લસ્ટર લોડિંગનો અનુભવ કરો છો ત્યારે તે કેસ માટે ઑટોસ્કેલર એ સારો અને અનુકૂળ ઉકેલ છે. તમે એકસાથે પીક લોડ માટે ઇચ્છિત ક્લસ્ટર ગોઠવણી હાંસલ કરો છો અને તે જ સમયે આ ક્લસ્ટરને અંડરલોડ દરમિયાન રાખશો નહીં, પૈસાની બચત કરો. સારું, વત્તા આ બધું તમારી સહભાગિતા વિના આપમેળે થાય છે. ઑટોસ્કેલર પોતે ક્લસ્ટર મેનેજર API અને ક્લાઉડ પ્રદાતા API ને વિનંતીઓના સમૂહ સિવાય બીજું કંઈ નથી, જે ચોક્કસ તર્ક અનુસાર લખાયેલ છે. તમારે જે ચોક્કસપણે યાદ રાખવાની જરૂર છે તે ગાંઠોનું 3 પ્રકારોમાં વિભાજન છે, જેમ આપણે અગાઉ લખ્યું હતું. અને તમે ખુશ થશો.

સોર્સ: www.habr.com