클러스터에 대한 자체 자동 크기 조정기를 만드는 방법

안녕하세요! 우리는 빅데이터를 다룰 수 있도록 사람들을 교육합니다. 모든 참가자가 함께 작업하는 자체 클러스터 없이 빅데이터에 대한 교육 프로그램을 상상하는 것은 불가능합니다. 이런 이유로 우리 프로그램에는 항상 그런 것이 있습니다 🙂 우리는 구성, 조정 및 관리에 참여하고 있으며 직원들은 그곳에서 직접 MapReduce 작업을 시작하고 Spark를 사용합니다.

이 게시물에서는 클라우드를 사용하여 자체 자동 크기 조정기를 작성하여 고르지 않은 클러스터 로딩 문제를 해결한 방법을 알려 드리겠습니다. Mail.ru 클라우드 솔루션.

문제

우리 클러스터는 일반 모드에서 사용되지 않습니다. 처리가 매우 고르지 않습니다. 예를 들어, 30명 전원과 선생님 한 명이 클러스터에 가서 사용하기 시작하는 실습 수업이 있습니다. 혹은 마감일을 앞두고 부하가 크게 늘어나는 날도 있습니다. 나머지 시간에는 클러스터가 언더로드 모드로 작동합니다.

솔루션 #1은 최대 부하를 견딜 수 있지만 나머지 시간에는 유휴 상태인 클러스터를 유지하는 것입니다.

솔루션 #2는 클래스 이전과 최대 로드 중에 노드를 수동으로 추가하는 작은 클러스터를 유지하는 것입니다.

솔루션 #3은 작은 클러스터를 유지하고 클러스터의 현재 로드를 모니터링하고 다양한 API를 사용하여 클러스터에서 노드를 추가 및 제거하는 자동 크기 조정기를 작성하는 것입니다.

이번 포스팅에서는 솔루션 #3에 대해 이야기하겠습니다. 이 자동 크기 조정기는 내부 요인보다는 외부 요인에 크게 의존하며 공급자가 이를 제공하지 않는 경우가 많습니다. 우리는 Mail.ru Cloud Solutions 클라우드 인프라를 사용하고 MCS API를 사용하여 자동 크기 조정기를 작성했습니다. 그리고 데이터 작업 방법을 가르치기 때문에 자신의 목적에 맞게 유사한 자동 크기 조정기를 작성하고 클라우드에서 사용할 수 있는 방법을 보여 주기로 결정했습니다.

사전 조건

먼저 Hadoop 클러스터가 있어야 합니다. 예를 들어 HDP 분포를 사용합니다.

노드를 신속하게 추가하고 제거하려면 노드 간에 일정한 역할 분배가 있어야 합니다.

1. 마스터 노드. 글쎄, 특별히 설명할 필요는 없습니다. 예를 들어 대화형 모드를 사용하는 경우 Spark 드라이버가 실행되는 클러스터의 기본 노드입니다.
2. 날짜 노드. 이는 HDFS에 데이터를 저장하고 계산이 수행되는 노드입니다.
3. 컴퓨팅 노드. 이는 HDFS에 아무 것도 저장하지 않지만 계산이 이루어지는 노드입니다.

중요한 점. 세 번째 유형의 노드로 인해 자동 확장이 발생합니다. 두 번째 유형의 노드를 가져와 추가하기 시작하면 응답 속도가 매우 느려집니다. 클러스터에서 해제하고 다시 커밋하는 데 몇 시간이 걸립니다. 물론 이는 자동 크기 조정에서 기대하는 것과는 다릅니다. 즉, 첫 번째 및 두 번째 유형의 노드를 건드리지 않습니다. 이는 프로그램 기간 동안 존재할 최소 실행 가능한 클러스터를 나타냅니다.

따라서 자동 크기 조정기는 Python 3으로 작성되었으며 Ambari API를 사용하여 클러스터 서비스를 관리하고 Mail.ru 클라우드 솔루션의 API (MCS) 기계 시동 및 정지용.

솔루션 아키텍처

1. 기준 치수 autoscaler.py. 여기에는 1) Ambari 작업을 위한 함수, 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 – Ambari의 클러스터 이름
• headers = {'X-Requested-By': 'ambari'}
• 그리고 내부 auth Ambari의 로그인 및 비밀번호는 다음과 같습니다. auth = ('login', 'password').

함수 자체는 REST API를 통해 Ambari에 대한 몇 번의 호출에 지나지 않습니다. 논리적인 관점에서 볼 때, 우리는 먼저 노드에서 실행 중인 서비스 목록을 받은 다음 특정 클러스터, 특정 노드에서 목록에서 상태로 서비스를 전송하도록 요청합니다. INSTALLED. 모든 서비스를 시작하고 노드를 상태로 전송하는 기능 Maintenance 등은 비슷해 보입니다. API를 통한 몇 가지 요청일 뿐입니다.

클래스 MC

클래스를 포함하는 코드는 다음과 같습니다. 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 클라우드 내부의 프로젝트 ID와 사용자 ID, 비밀번호를 전달합니다. 기능 중 vm_turn_on 기계 중 하나를 켜려고 합니다. 여기서의 논리는 조금 더 복잡합니다. 코드 시작 부분에서는 세 가지 다른 함수가 호출됩니다. 1) 토큰을 가져와야 합니다. 2) 호스트 이름을 MCS의 컴퓨터 이름으로 변환해야 하며 3) 이 컴퓨터의 ID를 가져와야 합니다. 다음으로 간단히 post 요청을 하고 이 머신을 실행합니다.

토큰을 얻는 함수는 다음과 같습니다.

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. 스크립트 관찰자.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)

여기에서 클러스터 용량을 늘리기 위한 조건이 생성되었는지, 예비 머신이 있는지 확인하고, 그 중 하나의 호스트 이름을 가져와 클러스터에 추가하고 이에 대한 메시지를 우리 팀의 Slack에 게시합니다. 그 후에는 시작됩니다 cooldown_period, 클러스터에 아무것도 추가하거나 제거하지 않고 단순히 로드를 모니터링하는 경우입니다. 안정화되어 최적의 부하 값 범위 내에 있으면 계속 모니터링하면 됩니다. 하나의 노드가 충분하지 않으면 다른 노드를 추가합니다.

앞으로 강의가 있는 경우 하나의 노드로는 충분하지 않다는 것을 이미 알고 있으므로 모든 무료 노드를 즉시 시작하고 강의가 끝날 때까지 활성 상태를 유지합니다. 이는 활동 타임스탬프 목록을 사용하여 발생합니다.

결론

Autoscaler는 클러스터 로딩이 고르지 않은 경우에 유용하고 편리한 솔루션입니다. 최대 로드에 대해 원하는 클러스터 구성을 동시에 달성하는 동시에 로드가 적은 동안에는 이 클러스터를 유지하지 않으므로 비용이 절약됩니다. 게다가 이 모든 일은 귀하의 참여 없이도 자동으로 발생합니다. 자동 크기 조정기 자체는 특정 논리에 따라 작성된 클러스터 관리자 API 및 클라우드 공급자 API에 대한 일련의 요청에 지나지 않습니다. 반드시 기억해야 할 것은 앞서 썼던 것처럼 노드를 3가지 유형으로 나누는 것입니다. 그리고 당신은 행복할 것입니다.

출처 : habr.com