YouDoでの機械学習がどのように実稼働に移行するか。 Yandexでの講義

大規模なサービスでは、機械学習を使用して問題を解決することは、作業の一部のみを実行することを意味します。 MLモデルの埋め込みはそれほど簡単ではなく、それらの周りにCI / CDプロセスを構築することはさらに困難です。 Yandexカンファレンス「Data＆Science：the application program」で、 YouDoのデータサイエンス責任者であるAdam Eldarovは、モデルのライフサイクルの管理、再トレーニングと再トレーニングプロセスのセットアップ、スケーラブルなマイクロサービスの開発などについて話しました。





-はじめに始めましょう。 データサイエンティストがいます。彼はJupyter Notebookにいくつかのコードを書いており、機能エンジニアリング、相互検証、モデルモデルのトレーニングを行っています。 速度は向上しています。







しかし、ある時点で彼は理解しています。会社のビジネス価値をもたらすために、彼はソリューションを生産のどこかで、神話の生産に結び付けなければなりません。 ほとんどの場合、本番環境で見たラップトップは送信できません。 そして、問題が発生します。ラップトップ内でこのコードを特定のサービスに出荷する方法です。 ほとんどの場合、APIを持つサービスを作成する必要があります。 または、PubSub、キューを介して通信します。







推奨事項を作成するとき、モデルをトレーニングして再トレーニングする必要があります。 このプロセスを監視する必要があります。 この場合、コード自体とモデルの両方を常にテストで確認する必要があります。そのため、ある時点でモデルが狂わず、つま先を予測し始めることはありません。 また、ABテストを通じて実際のユーザーで確認する必要があります。



どのようにコードにアプローチしますか？ GitLabがあります。 すべてのコードは、特定のドメインの問題を解決する多くの小さなライブラリに分割されています。 同時に、これは別個のGitLabプロジェクト、Gitバージョン管理、およびGitFlow分岐モデルです。 事前テストフックなどを使用して、統計テストチェックを満たさないコードをコミットできないようにします。 そして、テスト自体、ユニットテスト。 私たちは、プロパティベースのテストアプローチを使用します。







通常、テストを記述するとき、テスト関数と、手で作成する引数、いくつかの例、およびテスト関数が返す値を持っていることを意味します。 これは不便です。 コードは膨らんでいて、原則として多くの人はそれを書くのが面倒です。 その結果、テストによって明らかになったコードの束があります。 プロパティベースのテストは、すべての引数に特定の分布があることを意味します。 フェージングを行い、これらの分布からすべての引数をサンプリングし、これらの引数を使用してテスト対象の関数を呼び出し、特定のプロパティについてこの関数の結果を確認します。 その結果、コードがはるかに少なくなり、同時に多くのテストがあります。







GitFlowとは何ですか？ これは分岐モデルであり、2つの主要なブランチがあることを意味します-開発とマスター、製品の準備ができているコードがあり、すべての開発は開発ブランチで実行され、すべての新機能は機能ブランチから取得されます。 つまり、各機能は新しい機能ブランチであり、機能ブランチは短命でなければならず、機能トグルでもカバーされます。 その後、リリースを作成し、devからmasterに変更をスローし、ライブラリまたはサービスのバージョンタグをその上に配置します。







私たちは開発を行っており、いくつかの機能を見つけてGitLabにプッシュし、機能ブランチからバージンへのマージリクエストを作成しています。 トリガーが動作し、テストを実行します。すべてが正常であれば、フリーズできます。 しかし、それを保持しているのは私たちではなく、チームの誰かです。 コードを修正し、それによりバス係数を増加させます。 このコードセクションは既に2人に知られています。 その結果、誰かがバスにぶつかった場合、誰かはすでに自分が何をしているかを知っています。







ライブラリの継続的な統合は、通常、変更のテストのように見えます。 また、リリースすると、パッケージのプライベートPyPIサーバーにも公開されます。







さらに、パイプラインで収集できます。 これには、Luigiライブラリを使用します。 タスクなどのエンティティで動作します。タスクには出力があり、タスクの実行中に作成されたアーティファクトが保存されます。 実行するビジネスロジックをパラメーター化し、タスクとその出力を識別するタスクパラメーターがあります。 さらに、タスクには常に他のタスクが設定する要件があります。 ある種のタスクを実行すると、その出力をチェックすることですべての依存関係がチェックされます。 出力が存在する場合、依存関係は開始されません。 一部のストレージからアーティファクトが欠落している場合、開始されます。 これにより、パイプライン、有向巡回グラフが形成されます。







すべてのパラメーターはビジネスロジックを識別します。 そうすることで、彼らはアーティファクトを特定します。 これは常に、ある種の粒度、感度、または週、日、時間、3時間の日付です。 モデルをトレーニングすると、Luigi taskaには常にこのタスクのハイパーパラメーターがあり、それらが生成するアーティファクトに漏れ、ハイパーパラメーターがアーティファクトの名前に反映されます。 したがって、基本的にすべての中間データセットと最終成果物をバージョン管理し、それらは上書きされることはなく、常にストレージにのみ更新され、ストレージは何らかの種類のピクニック、モデルなどの最終成果物を見るHDFSとS3プライベートを実行します。 そして、すべてのパイプラインコードは、関連するリポジトリのサービスプロジェクトにあります。







なんとか修正する必要があります。 HashiCorpスタックが役に立ちます。Terraformを使用してコードの形でインフラストラクチャを宣言し、Vaultで秘密を管理します。すべてのパスワード、データベースへの外観があります。 Consulは、構成に使用できるキー値ストレージによって配布される検出サービスです。 また、Consulはノードとサービスのヘルスチェックを行い、可用性をチェックします。



そして-遊牧民。 それはオーケストレーションのシステムであり、サービスとある種のバッチジョブを排除します。







どのように使用しますか？ Luigiパイプラインがあります。Dockerコンテナーにパックし、バットまたは定期的なバッチジョブをNomadにドロップします。 バッチジョブ-これは完了したものであり、すべてが成功した場合-すべてがOKであれば、手動で再開できます。 しかし、何か問題が発生した場合、Nomadは試行を使い果たすか、正常に終了しないまで再試行します。



定期的なバッチジョブ-これはまったく同じで、スケジュールに従ってのみ機能します。



問題があります。 何らかのオーケストレーションシステムにコンテナを展開する場合、このコンテナ、CPU、またはメモリが必要とするメモリ量を示す必要があります。 3時間実行するパイプラインがある場合、2時間で10 GBのRAM、1時間-70 GBが消費されます。 彼に与えた制限を超えると、Dockerデーモンが来てDockersを殺します（。）[02:26:13]絶えずメモリを追い出したくないので、70 GBすべてのピークメモリ負荷を指定する必要があります。 しかし、ここに問題があります。3時間の70 GBがすべて割り当てられ、他のジョブにはアクセスできなくなります。



したがって、私たちは他の方法で行った。 私たちのLuigiパイプライン全体は、それ自体ではビジネスロジックの種類を開始せず、Nomad、いわゆるパラメーター化されたジョブでサイコロのセットを起動するだけです。 実際、これはサーバー（nrzb）[02:26:39]機能の類似物であり、AVS Lambdaは知っています。 ライブラリを作成するとき、パラメータ化されたジョブ、つまりいくつかのパラメータを持つコンテナの形式ですべてのコードをCIを介して展開します。 たとえば、Lite JBM Classifierには、トレーニング用の入力データへのパスへのパラメーター、モデルのハイパーパラメーター、および出力アーティファクトへのパスがあります。 これらはすべてNomadに登録されており、LuigiパイプラインからAPIを介してこれらすべてのNomadジョブをプルできますが、Luigiは同じタスクを何度も実行しないようにします。



常に同じテキスト処理を行っているとしましょう。 10個の条件付きモデルがあり、毎回テキスト処理を再開したくありません。 これは一度だけ開始され、同時に再利用されるたびに完成した結果になります。 同時に、これらはすべて分散的に機能します。大きなクラスターで巨大なグリッド検索を実行でき、鉄を捨てる時間しかありません。







アーティファクトがあります。何らかの形でこれをサービスの形に整える必要があります。 サービスは、HTTP APIを公開するか、キューを介して通信します。 この例では、これはHTTP APIであり、最も単純な例です。 同時に、サービスとの通信、またはHTTP JSON APIを介して他のサービスと通信するサービスは、JSONスキームを検証します。 サービス自体は、APIおよびこのオブジェクトのスキームのドキュメントで常にJSONオブジェクトを記述します。 ただし、すべてのJSONオブジェクトフィールドが常に必要なわけではないため、消費者主導のコントラクトが検証され、このスキームが検証され、カスケード型障害による分散システムの障害を防ぐために、パターンサーキットブレーカーを介して通信が行われます。



同時に、Consulがこのサービスの可用性を確認できるように、サービスはHTTPヘルスチェックを設定する必要があります。 同時に、Nomadは、3つのhelloチェックのサービスが連続して存在するようにすることができ、サービスを再起動して支援することができます。 サービスはすべてのログをJSON形式で書き込みます。 JSONロギングドライバーとElasticsスタックを使用します。FileBitはすべての時点ですべてのJSONログを取得してログキャッシュに入れ、そこからElasticに到達すると、KBanを分析できます。 同時に、メトリックを収集してダッシュボードを構築するためにログを使用しません。それは非効率的です。これにはPrometheusを使用するシステムを使用します。各ダッシュボードサービスのテンプレートを作成するプロセスがあり、サービスによって生成される技術メトリックを分析できます



さらに、何か問題が発生した場合、アラートが入りますが、ほとんどの場合、これでは十分ではありません。 セントリーは私たちの助けになります。これはインシデント分析のためのものです。 実際、すべてのエラーレベルログをSentryハンドラーでキャプチャし、Sentryにプッシュします。 そして、詳細なトレースバックがあります。サービスがどの環境にあったか、どのバージョン、どの関数がどのスコープで呼び出されたか、このスコープ内のどの変数がどの値であったかに関するすべての情報があります。 すべての設定、これはすべて表示され、何が起こったのかを素早く理解してエラーを修正するのに役立ちます。







その結果、サービスは次のようになります。 別のGitLabプロジェクト、パイプラインコード、テストコード、サービスコード自体、さまざまな構成、Nomad、CI構成、APIドキュメント、コミットフックなど。







リリースを行うとき、次のようにCIを実行します：コンテナーを構築し、テストを実行し、ステージングにクラスターをドロップし、そこでサービスのテストコントラクトを実行し、負荷テストを実施して、予測が遅すぎないことを確認し、考えている負荷を維持します。 すべてが問題なければ、このサービスを実稼働環境に展開します。 そして、2つの方法があります：定期的なバッチジョブがバックグラウンドで動作してアーティファクトを生成する場合、またはパイプを使用して何らかのパイプラインをトリガーする場合、パイプラインを展開することができます。サービスを展開します。







この場合、他に何が起こりますか？ 機能ブランチの開発には、機能トグルのようなパラダイムがあると言いました。 良い方法では、何かがうまくいかなかった場合に戦闘で機能をノックアウトするために、いくつかのトグルで機能をカバーする必要があります。 その後、リリーストレインですべての機能を収集し、機能が未完成であっても展開できます。 機能の切り替えだけがオフになります。 私たちは全員データサイエンティストなので、AVテストも行いたいと考えています。 LightGBMをCatBoostに置き換えたとしましょう。 これを確認したいのですが、同時に、AVテストはユーザーIDを参照して管理されます。 機能トグルはuserIDにバインドされているため、AVテストに合格します。 ここでこれらのメトリックを確認する必要があります。



すべてのサービスはNomadに展開されます。 2つのNomadプロダクションクラスターがあります。1つはバッチジョブ用で、もう1つはサービス用です。







すべてのビジネスイベントをKafkaにプッシュします。 そこから拾うことができます。 本質的に、それは子羊のアーキテクチャです。 いくつかのサービスでHDFSにサブスクライブし、リアルタイム分析を行い、同時にClickHouseですべてをすくい取り、ダッシュボードを構築してサービスのすべてのビジネスイベントを分析できます。 AVテストを分析できます。







また、コードを変更しなかった場合は、機能の切り替えを使用しないでください。 ペンでパイプラインを突っ込んだだけで、彼は新しいモデルを教えてくれました。 それへの新しい道があります。 構成内のモデルへのNomadパスを変更し、新しいサービスをリリースするだけです。ここでCanary Deploymentパラダイムが助けになります。Nomadでボックスから利用できます。



現在のバージョンのサービスは3つのインスタンスにあります。 3つのカナリアが必要だと言います-古いバージョンを削減することなく、新しいバージョンのレプリカをさらに3つ展開します。 その結果、トラフィックは2つの部分に分割され始めます。 トラフィックの一部は、サービスの新しいバージョンに分類されます。 すべてのサービスは、すべてのビジネスイベントをKafkaにプッシュします。 その結果、メトリックをリアルタイムで分析できます。



すべてが正常であれば、すべてが正常であると言えます。 デプロイすると、Nomadはすべての古いバージョンを静かにオフにし、新しいバージョンをスケーリングします。



このモデルは、ユーザーアイテムというエンティティによってバージョンルーティングをバインドする必要がある場合に問題があります。 トラフィックはラウンドロビンでバランスが取られるため、このようなスキームは機能しません。 そのため、次の方法でサービスを2つの部分に分けました。







これは、ゲートウェイ層とワーカー層です。 クライアントは、HTTPを介してゲートウェイレイヤーと通信します。バージョンを選択し、トラフィックを分散するためのすべてのロジックはゲートウェイにあります。 同時に、述部を完了するために必要なすべてのI / Oバウンドタスクもゲートウェイに配置されます。 リクエストの述語でuserIDを取得し、何らかの情報を追加する必要があるとします。 他のマイクロサービスをプルして、すべての情報、機能、またはベースを取得する必要があります。 その結果、これらはすべてゲートウェイで発生します。 彼はモデルにのみいる労働者と通信し、1つのことを行います-予測します。 入力および出力。



しかし、サービスを2つの部分に分割したため、リモートネットワーク呼び出しのためにオーバーヘッドが発生しました。 それをレベル化する方法は？ HTTP2の上で実行されるGoogleのJRPCフレームワーク、GoogleのRPCが助けになります。 多重化と圧縮を使用できます。 JPRCはプロトバフを使用します。 これは、高速シリアル化と逆シリアル化を行う、強く型付けされたバイナリプロトコルです。



その結果、ゲートウェイとワーカーを個別にスケーリングすることもできます。 一定量のオープンHTTP接続を維持できないとしましょう。 さて、ゲートウェイをスケーリングします。 予測が遅すぎて、負荷を維持する時間がありません。 このアプローチは、多腕バンディットに非常に適しています。 ゲートウェイでは、すべてのトラフィックバランシングロジックが実装されているため、外部マイクロサービスにアクセスし、そこから各バージョンのすべての統計を取得し、トラフィックのバランスをとる方法を決定できます。 トンプソンサンプリングを使用するとしましょう。







すべて問題ありません。モデルは何らかの形で訓練されたので、Nomad configに登録しました。 しかし、トレーニング中に陳腐化する時間がすでにある推奨のモデルがあり、それらを絶えず再トレーニングする必要がある場合はどうでしょうか？ すべてが同じ方法で行われます。定期的なバッチジョブによって、たとえば3時間ごとに何らかのアーティファクトが生成されます。 同時に、彼の仕事の終わりに、パイプラインはConsulの新しいモデルのパスを設定します。 これはキー値ストレージであり、構成に使用されます。 Nomadは構成を構成できます。 キー値ストレージConsulの値に基づいた環境変数があるとします。 彼は変更を監視し、新しいパスが表示されるとすぐに、2つのパスを取ることができると判断します。 彼は、新しいリンクを介してアーティファクト自体をダウンロードし、ボリュームとリブートを使用してサービスコンテナーをDockerに配置します。これをすべて実行して、ダウンタイムが発生しないようにします。 または、彼は新しい設定をレンダリングし、サービスを彼に報告します。 または、サービス自体がそれを検出します-そして、それ自体は独立して、そのモデカをライブ更新できます。 それだけです、ありがとう。