🥇 क्लस्टरसाठी तुमचा स्वतःचा ऑटोस्केलर कसा बनवायचा

नमस्कार! आम्ही लोकांना मोठ्या डेटासह काम करण्याचे प्रशिक्षण देतो. मोठ्या डेटावरील शैक्षणिक कार्यक्रमाची त्याच्या स्वतःच्या क्लस्टरशिवाय कल्पना करणे अशक्य आहे, ज्यावर सर्व सहभागी एकत्र काम करतात. या कारणास्तव, आमच्या प्रोग्राममध्ये ते नेहमीच असते :) आम्ही त्याचे कॉन्फिगरेशन, ट्यूनिंग आणि प्रशासनामध्ये गुंतलेले असतो आणि लोक थेट मॅपरेड्यूस जॉब्स तेथे लॉन्च करतात आणि स्पार्क वापरतात.

या पोस्टमध्ये आम्ही तुम्हाला क्लाउड वापरून आमचे स्वतःचे ऑटोस्केलर लिहून असमान क्लस्टर लोडिंगची समस्या कशी सोडवली ते सांगू. Mail.ru क्लाउड सोल्यूशन्स.

समस्या

आमचा क्लस्टर ठराविक मोडमध्ये वापरला जात नाही. विल्हेवाट अत्यंत असमान आहे. उदाहरणार्थ, व्यावहारिक वर्ग आहेत, जेव्हा सर्व 30 लोक आणि एक शिक्षक क्लस्टरमध्ये जातात आणि ते वापरण्यास प्रारंभ करतात. किंवा पुन्हा, अंतिम मुदतीपूर्वीचे दिवस आहेत जेव्हा लोड मोठ्या प्रमाणात वाढते. उर्वरित वेळ क्लस्टर अंडरलोड मोडमध्ये कार्य करतो.

उपाय #1 म्हणजे एक क्लस्टर ठेवणे जे पीक भार सहन करेल, परंतु उर्वरित वेळेत निष्क्रिय असेल.

उपाय #2 म्हणजे एक लहान क्लस्टर ठेवणे, ज्यामध्ये तुम्ही वर्गांपूर्वी आणि पीक लोड दरम्यान नोड्स मॅन्युअली जोडता.

उपाय #3 म्हणजे एक लहान क्लस्टर ठेवणे आणि एक ऑटोस्केलर लिहिणे जे क्लस्टरच्या वर्तमान लोडचे निरीक्षण करेल आणि विविध API वापरून, क्लस्टरमधून नोड्स जोडणे आणि काढून टाकणे.

या पोस्टमध्ये आपण उपाय #3 बद्दल बोलू. हे ऑटोस्केलर अंतर्गत घटकांऐवजी बाह्य घटकांवर जास्त अवलंबून असते आणि प्रदाते सहसा ते प्रदान करत नाहीत. आम्ही Mail.ru Cloud Solutions क्लाउड इन्फ्रास्ट्रक्चर वापरतो आणि MCS API वापरून ऑटोस्केलर लिहितो. आणि आम्ही डेटासह कसे कार्य करावे हे शिकवल्यामुळे, आम्ही तुमच्या स्वत: च्या हेतूंसाठी समान ऑटोस्केलर कसे लिहू शकता आणि ते तुमच्या क्लाउडसह कसे वापरू शकता हे दर्शविण्याचे ठरवले आहे.

पूर्वापेक्षित

प्रथम, तुमच्याकडे हडूप क्लस्टर असणे आवश्यक आहे. उदाहरणार्थ, आम्ही एचडीपी वितरण वापरतो.

तुमचे नोड्स द्रुतपणे जोडले जाण्यासाठी आणि काढून टाकण्यासाठी, तुमच्याकडे नोड्समध्ये भूमिकांचे विशिष्ट वितरण असणे आवश्यक आहे.

मास्टर नोड. बरं, येथे स्पष्ट करण्यासाठी विशेषतः आवश्यक काहीही नाही: क्लस्टरचा मुख्य नोड, ज्यावर, उदाहरणार्थ, स्पार्क ड्राइव्हर लॉन्च केला जातो, जर तुम्ही परस्परसंवादी मोड वापरत असाल.
तारीख नोड. हा नोड आहे ज्यावर तुम्ही HDFS वर डेटा संग्रहित करता आणि जिथे गणना केली जाते.
संगणकीय नोड. हा एक नोड आहे जिथे तुम्ही HDFS वर काहीही संचयित करत नाही, परंतु जिथे गणना केली जाते.

महत्त्वाचा मुद्दा. तिसऱ्या प्रकारच्या नोड्समुळे ऑटोस्केलिंग होईल. जर तुम्ही दुसऱ्या प्रकारचे नोड्स घेणे आणि जोडणे सुरू केले, तर प्रतिसादाची गती खूपच कमी असेल - डिकमिशनिंग आणि रिकममिटिंगला तुमच्या क्लस्टरवर काही तास लागतील. हे, अर्थातच, ऑटोस्केलिंगकडून तुम्हाला अपेक्षित नाही. म्हणजेच, आम्ही पहिल्या आणि द्वितीय प्रकारच्या नोड्सला स्पर्श करत नाही. ते किमान व्यवहार्य क्लस्टरचे प्रतिनिधित्व करतील जे कार्यक्रमाच्या संपूर्ण कालावधीत अस्तित्वात असतील.

तर, आमचे ऑटोस्केलर पायथन 3 मध्ये लिहिलेले आहे, क्लस्टर सेवा व्यवस्थापित करण्यासाठी अंबारी API वापरते, वापरते Mail.ru क्लाउड सोल्यूशन्स कडून API (MCS) मशीन सुरू करण्यासाठी आणि थांबवण्यासाठी.

समाधान आर्किटेक्चर

मॉड्यूल autoscaler.py. यात तीन वर्ग आहेत: 1) अंबारीसह कार्य करण्यासाठी कार्ये, 2) MCS सह कार्य करण्यासाठी कार्ये, 3) ऑटोस्केलरच्या तर्काशी थेट संबंधित कार्ये.
स्क्रिप्ट observer.py. मूलत: यात भिन्न नियम असतात: ऑटोस्केलर फंक्शन्स कधी आणि कोणत्या क्षणी कॉल करायचे.
कॉन्फिगरेशन फाइल 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, स्केलिंगसाठी परवानगी असलेल्या नोड्सची सूची, तसेच नोड कॉन्फिगरेशन पॅरामीटर्स: YARN मधील नोडला वाटप केलेली मेमरी आणि cpu. 2 अंतर्गत पॅरामीटर्स q_ram, q_cpu देखील आहेत, जे रांग आहेत. त्यांचा वापर करून, आम्ही सध्याच्या क्लस्टर लोडची मूल्ये संग्रहित करतो. गेल्या 5 मिनिटांत सातत्याने लोड वाढत असल्याचे आपण पाहिल्यास, आपण क्लस्टरमध्ये +1 नोड जोडणे आवश्यक आहे असे ठरवू. क्लस्टर अंडरयुटिलायझेशन स्थितीसाठी हेच खरे आहे.

वरील कोड फंक्शनचे उदाहरण आहे जे क्लस्टरमधून मशीन काढून टाकते आणि क्लाउडमध्ये थांबवते. प्रथम डिकमिशनिंग आहे YARN Nodemanager, नंतर मोड चालू होईल Maintenance, नंतर आम्ही मशीनवरील सर्व सेवा थांबवतो आणि क्लाउडमधील व्हर्च्युअल मशीन बंद करतो.

2. Script observer.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

क्लस्टरसाठी तुमचा स्वतःचा ऑटोस्केलर कसा बनवायचा