ڪلستر لاءِ پنهنجو آٽو اسڪيلر ڪيئن ٺاهيو

سلام! اسان ماڻهن کي وڏي ڊيٽا سان ڪم ڪرڻ جي تربيت ڏيون ٿا. اهو تصور ڪرڻ ناممڪن آهي ته هڪ تعليمي پروگرام وڏي ڊيٽا تي پنهنجي ڪلستر کان سواء، جنهن تي سڀني شرڪت گڏجي ڪم ڪن. ان لاءِ، اسان جي پروگرام ۾ هميشه اهو هوندو آهي :) اسان ان جي ترتيب، ٽيوننگ ۽ ايڊمنسٽريشن ۾ مصروف آهيون، ۽ ماڻهو سڌو سنئون اتي MapReduce نوڪريون شروع ڪن ٿا ۽ اسپارڪ استعمال ڪن ٿا.

هن پوسٽ ۾ اسين توهان کي ٻڌائي سگهنداسين ته اسان ڪلستر جي اڻ برابري واري لوڊشيڊنگ جي مسئلي کي ڪيئن حل ڪيو ڪلائوڊ استعمال ڪندي پنهنجو پنهنجو آٽو اسڪيلر لکي Mail.ru ڪلائوڊ حل.

مسئلو

اسان جو ڪلستر هڪ عام موڊ ۾ استعمال نه ڪيو ويو آهي. نيڪال انتهائي ناياب آهي. مثال طور، عملي ڪلاس آهن، جڏهن سڀ 30 ماڻهو ۽ هڪ استاد ڪلستر ڏانهن وڃو ۽ ان کي استعمال ڪرڻ شروع ڪيو. يا ٻيهر، آخري وقت کان پهريان ڏينهن آهن جڏهن لوڊ تمام گهڻو وڌي ٿو. باقي وقت ڪلستر انڊر لوڊ موڊ ۾ هلندي آهي.

حل # 1 ھڪڙو ڪلستر رکڻو آھي جيڪو چوٽي جي لوڊ کي برداشت ڪندو، پر باقي وقت بيڪار رھندو.

حل # 2 ھڪڙو ننڍڙو ڪلستر رکڻ آھي، جنھن ۾ توھان دستي طور تي نوڊس شامل ڪريو ڪلاس کان اڳ ۽ چوٽي لوڊ دوران.

حل #3 هڪ ننڍڙو ڪلستر رکڻ ۽ هڪ آٽو اسڪيلر لکڻ آهي جيڪو ڪلستر جي موجوده لوڊ مانيٽر ڪندو ۽، مختلف APIs استعمال ڪندي، ڪلستر مان نوڊس شامل ۽ هٽائي ڇڏيندو.

هن پوسٽ ۾ اسان حل #3 بابت ڳالهائينداسين. هي آٽو اسڪيلر اندروني عنصرن جي بجاءِ ٻاهرين عنصرن تي تمام گهڻو ڀاڙيندو آهي، ۽ مهيا ڪندڙ اڪثر ان کي مهيا نه ڪندا آهن. اسان Mail.ru Cloud Solutions ڪلائوڊ انفراسٽرڪچر استعمال ڪندا آهيون ۽ MCS API استعمال ڪندي هڪ آٽو اسڪيلر لکيو آهي. ۽ جيئن ته اسان سيکاريندا آهيون ته ڊيٽا سان ڪيئن ڪم ڪجي، اسان اهو ڏيکارڻ جو فيصلو ڪيو ته توهان پنهنجي مقصدن لاءِ ساڳيو آٽو اسڪيلر ڪيئن لکي سگهو ٿا ۽ ان کي پنهنجي ڪلائوڊ سان استعمال ڪري سگهو ٿا.

تعارف

پهرين، توهان کي هڪ Hadoop ڪلستر هجڻ گهرجي. مثال طور، اسان استعمال ڪندا آهيون HDP تقسيم.

توھان جي نوڊس کي جلدي شامل ڪرڻ ۽ ختم ڪرڻ لاءِ، توھان کي نوڊس جي وچ ۾ ڪردار جي ھڪڙي خاص تقسيم ٿيڻ گھرجي.

1. ماسٽر نوڊ. خير، هتي وضاحت ڪرڻ لاءِ خاص طور تي ڪجهه به ضروري ناهي: ڪلستر جو بنيادي نوڊ، جنهن تي، مثال طور، اسپارڪ ڊرائيور شروع ڪيو ويو آهي، جيڪڏهن توهان انٽرايڪٽو موڊ استعمال ڪندا آهيو.
2. تاريخ نوڊ. هي اهو نوڊ آهي جنهن تي توهان HDFS تي ڊيٽا محفوظ ڪريو ٿا ۽ جتي حساب ڪتاب ٿين ٿا.
3. ڪمپيوٽنگ نوڊ. هي هڪ نوڊ آهي جتي توهان HDFS تي ڪجھ به ذخيرو نٿا ڪريو، پر جتي حساب ڪتاب ٿئي ٿي.

اهم نقطو. ٽين قسم جي نوڊس جي ڪري خودڪار اسڪيلنگ ٿيندي. جيڪڏهن توهان ٻئي قسم جا نوڊس وٺڻ ۽ شامل ڪرڻ شروع ڪيو ٿا، ته جواب جي رفتار تمام گهٽ ٿي ويندي - ختم ڪرڻ ۽ ٻيهر ڪم ڪرڻ ۾ توهان جي ڪلستر تي ڪلاڪ لڳندا. اهو، يقينا، اهو ناهي جيڪو توهان آٽو اسڪيلنگ کان توقع ڪندا آهيو. اهو آهي، اسان پهرين ۽ ٻئي قسم جي نوڊس کي هٿ نه ڪندا آهيون. اهي گهٽ ۾ گهٽ قابل عمل ڪلستر جي نمائندگي ڪندا جيڪي پروگرام جي سڄي عرصي دوران موجود هوندا.

تنهن ڪري، اسان جو آٽو اسڪيلر Python 3 ۾ لکيل آهي، ڪلستر سروسز کي منظم ڪرڻ لاءِ Ambari API استعمال ڪري ٿو، استعمال ڪري ٿو API کان Mail.ru Cloud Solutions (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 ھي آھي توھان جو يوزر نالو ۽ پاسورڊ Ambari لاءِ: auth = ('login', 'password').

فنڪشن بذات خود امبري تائين REST API ذريعي ڪجهه ڪالن کان وڌيڪ ناهي. منطقي نقطي نظر کان، اسان پهريون ڀيرو هڪ نوڊ تي هلندڙ خدمتن جي هڪ فهرست حاصل ڪندا آهيون، ۽ پوء هڪ ڏنل ڪلستر تي، ڏنل نوڊ تي، فهرست مان رياست ڏانهن خدمتن کي منتقل ڪرڻ لاء. INSTALLED. سڀني خدمتن کي شروع ڪرڻ لاء ڪم، رياست ڏانهن نوڊس جي منتقلي لاء Maintenance وغيره هڪجهڙا نظر اچن ٿا - اهي صرف ڪجهه درخواستون آهن API ذريعي.

ڪلاس ميڪس

هي اهو آهي جيڪو ڪوڊ جو هڪ ٽڪرو جنهن ۾ هڪ ڪلاس نظر اچي ٿو 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 نوڊ شامل ڪرڻ جي ضرورت آهي. ساڳيو ئي ڪلستر جي غير استعمال واري رياست لاءِ صحيح آهي.

مٿي ڏنل ڪوڊ هڪ فنڪشن جو هڪ مثال آهي جيڪو ڪلستر مان هڪ مشين کي هٽائي ٿو ۽ ان کي بادل ۾ روڪي ٿو. پهرين اتي هڪ decommissioning آهي YARN Nodemanager، پوءِ موڊ آن ٿيندو Maintenance، پوءِ اسان مشين تي سڀئي خدمتون بند ڪريون ۽ ڪلائوڊ ۾ ورچوئل مشين کي بند ڪريون.

2. اسڪرپٽ 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، جڏهن اسان ڪلستر مان ڪجھ به شامل يا ختم نه ڪندا آهيون، پر صرف لوڊ جي نگراني ڪندا آهيون. جيڪڏهن اهو مستحڪم ٿي چڪو آهي ۽ بهتر لوڊ ويلز جي ڪوريڊور ۾ آهي، ته پوءِ اسان صرف نگراني جاري رکون ٿا. جيڪڏهن هڪ نوڊ ڪافي نه هو، پوء اسان هڪ ٻيو شامل ڪيو.

ڪيسن لاءِ جڏهن اسان وٽ سبق اڳيان آهي، اسان اڳ ۾ ئي پڪ ڄاڻون ٿا ته هڪ نوڊ ڪافي نه هوندو، تنهنڪري اسان فوري طور تي سڀئي مفت نوڊس شروع ڪريون ٿا ۽ سبق جي پڄاڻي تائين انهن کي فعال رکون ٿا. اهو ٿئي ٿو سرگرمي ٽائم اسٽيمپ جي فهرست استعمال ڪندي.

ٿڪل

Autoscaler انهن ڪيسن لاءِ هڪ سٺو ۽ آسان حل آهي جڏهن توهان اڻ برابري ڪلسٽر لوڊشيڊنگ جو تجربو ڪريو ٿا. توهان هڪ ئي وقت چوٽي جي لوڊ لاءِ گهربل ڪلسٽر ترتيب حاصل ڪريو ۽ ساڳئي وقت هن ڪلسٽر کي انڊر لوڊ دوران نه رکو، پئسا بچائي. خير، اهو سڀ ڪجهه توهان جي شموليت کان سواءِ پاڻمرادو ٿئي ٿو. خود اسڪيلر خود ڪلستر مئنيجر API ۽ ڪلائوڊ فراهم ڪندڙ API ڏانهن درخواستن جي هڪ سيٽ کان وڌيڪ ڪجهه ناهي، هڪ خاص منطق جي مطابق لکيو ويو آهي. ڇا توهان کي ضرور ياد رکڻ جي ضرورت آهي نوڊس جي تقسيم 3 قسمن ۾، جيئن اسان اڳ ۾ لکيو آهي. ۽ تون خوش ٿي ويندين.

جو ذريعو: www.habr.com