クラスター用に独自のオートスケーラーを作成する方法

こんにちは！ 私たちはビッグデータを扱う人材を訓練します。 すべての参加者が協力して作業する独自のクラスターがなければ、ビッグデータに関する教育プログラムを想像することは不可能です。 このため、私たちのプログラムには常にそれが含まれています 🙂 私たちはその構成、チューニング、管理に従事しており、スタッフはそこで MapReduce ジョブを直接起動し、Spark を使用しています。

この投稿では、クラウドを使用して独自のオートスケーラーを作成することで、クラスターの負荷が不均一になる問題をどのように解決したかについて説明します。 Mail.ru クラウド ソリューション.

問題

私たちのクラスターは通常のモードでは使用されません。 処分は非常に不均一です。 例えば、30人全員と先生がクラスターに行って使い始める実践的な授業があります。 あるいは、締め切り前に負荷が大幅に増加する日もあります。 残りの時間は、クラスターはアンダーロード モードで動作します。

解決策 #1 は、ピーク負荷には耐えられるが、残りの時間はアイドル状態になるクラスターを維持することです。

解決策 2 は、小規模なクラスターを維持し、クラスの前およびピーク負荷中に手動でノードを追加することです。

解決策 #3 は、小規模なクラスターを維持し、クラスターの現在の負荷を監視し、さまざまな API を使用してクラスターにノードを追加および削除するオートスケーラーを作成することです。

この投稿では、解決策 #3 について説明します。 このオートスケーラーは内部要因ではなく外部要因に大きく依存しており、プロバイダーが提供していないことがよくあります。 私たちは Mail.ru Cloud Solutions のクラウド インフラストラクチャを使用し、MCS API を使用してオートスケーラーを作成しました。 そして、私たちはデータの操作方法を教えているので、独自の目的のために同様のオートスケーラーを作成し、それをクラウドで使用する方法を示すことにしました。

前提条件

まず、Hadoop クラスターが必要です。 たとえば、HDP ディストリビューションを使用します。

ノードを迅速に追加および削除するには、ノード間で役割を一定に分散する必要があります。

1. マスターノード。 そうですね、特に説明する必要はありません。クラスターのメイン ノード。対話モードを使用する場合、たとえば Spark ドライバーが起動されます。
2. 日付ノード。 これは、HDFS にデータを保存し、計算が行われるノードです。
3. コンピューティングノード。 これは、HDFS に何も保存しないノードですが、計算が行われるノードです。

大事なポイント。 XNUMX 番目のタイプのノードにより自動スケーリングが発生します。 XNUMX 番目のタイプのノードの取得と追加を開始すると、応答速度が非常に遅くなり、クラスターでの廃止と再コミットに数時間かかります。 もちろん、これは自動スケーリングに期待されるものではありません。 つまり、XNUMX 番目と XNUMX 番目のタイプのノードには触れません。 これらは、プログラムの期間全体を通じて存在する、実行可能な最小限のクラスターを表します。

したがって、私たちのオートスケーラーは Python 3 で書かれており、Ambari API を使用してクラスター サービスを管理し、 Mail.ru クラウド ソリューションの API (MCS) マシンの起動と停止用。

ソリューションアーキテクチャ

1. モジュール autoscaler.py。 これには 1 つのクラスが含まれています: 2) Ambari で動作する関数、3) MCS で動作する関数、XNUMX) オートスケーラーのロジックに直接関連する関数。
2. スクリプト observer.py。 基本的に、オートスケーラー関数をいつ、どの瞬間に呼び出すかというさまざまなルールで構成されます。
3. 設定ファイル config.py。 たとえば、自動スケーリングが許可されているノードのリストや、新しいノードが追加された瞬間から待機する時間などに影響するその他のパラメーターが含まれています。 クラスの開始のタイムスタンプもあるため、クラスの前に最大許容クラスター構成が起動されます。

最初の XNUMX つのファイル内のコード部分を見てみましょう。

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 を介したいくつかのリクエストにすぎません。

クラスマック

クラスを含むコードは次のようになります 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 つの関数が呼び出されます: 3) トークンを取得する必要がある、XNUMX) ホスト名を MCS 内のマシンの名前に変換する必要がある、XNUMX) このマシンの ID を取得する。 次に、ポストリクエストを作成して、このマシンを起動します。

トークンを取得する関数は次のようになります。

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. スクリプト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)

その中で、クラスターの容量を増やすための条件が作成されているかどうか、および予備のマシンがあるかどうかを確認し、そのうちの XNUMX つのホスト名を取得してクラスターに追加し、それに関するメッセージをチームの Slack で公開します。 その後始まります cooldown_periodクラスターに何も追加または削除せず、単に負荷を監視する場合です。 安定し、最適な負荷値の範囲内にある場合は、監視を継続するだけです。 XNUMX つのノードでは不十分な場合は、別のノードを追加します。

レッスンが予定されている場合、XNUMX つのノードでは十分ではないことがすでにわかっているため、空いているすべてのノードをすぐに起動し、レッスンが終了するまでアクティブなままにします。 これは、アクティビティのタイムスタンプのリストを使用して行われます。

まとめ

オートスケーラーは、クラスターの負荷が不均一になる場合に適した便利なソリューションです。 ピーク負荷時に必要なクラスター構成を実現すると同時に、低負荷時にこのクラスターを維持しないことでコストを節約できます。 さらに、これはすべて、ユーザーの参加なしで自動的に行われます。 オートスケーラー自体は、特定のロジックに従って記述された、クラスター マネージャー API およびクラウド プロバイダー API への一連のリクエストにすぎません。 必ず覚えておきたいのは、先ほども書いたようにノードを3種類に分けることです。 そしてあなたは幸せになるでしょう。

出所： habr.com