みなさんこんにちは！



今日は、ストリーミングデータ処理を使用して、レコメンデーションの品質を向上させ、レコメンデーションシステム全体の応答時間を5倍短縮する方法について説明します。 それは、顧客の1人であるRutubeビデオストリーミングサービスです。







最初に、Rutube自体についてのいくつかの言葉と、それが推奨事項を必要とする理由。 まず、執筆時点で、当社の推奨システムは、51.76百万のユーザーと126万のアイテム、つまりビデオに関するデータを保存しています。 当然、近い将来、すべてのビデオを視聴できるユーザーはいなくなります。 推薦システムが彼の助けになります。 第二に、Rutubeは広告を表示することでお金を稼ぎます。 会社の主要なビジネス指標の1つは、サイトでユーザーが費やした時間です。 Rutubeはこのレートを最大化します。 推奨システムはこれを行うのに役立ちます。 そして第三に、Rutube自体の機能です。 コンテンツの一部は商用ビデオの著作権所有者です。 たとえば、TNTには独自のチャネルがあり、プログラムの問題を定期的に投稿しています。 「House-2」または「Dances」の新刊が出ると、人々は急いでそれらを見ます。 ビデオは本質的にバイラルです。 推奨システムは、そのようなビデオを追跡して他のユーザーに表示するのに役立ちます。

持っていたもの

2015年5月にE-ContentaでRutubeの推奨システムの最初のバージョンを作成しました。これは、次の項目で構成されました。n回ごとに推奨を再計算するアイテムベースの協調フィルタリング（ nはフィボナッチ数列の数） つまり、各ビデオの視聴はイベントであり、このイベントのシーケンス番号がフィボナッチシリーズに含まれていた場合、このビデオと他のビデオの近接度を再計算しました。 これは、 Tornado + Celery （ブローカーはRabbitMQ ）+ MongoDBのスタックを使用して実装されました。 つまり、データはTornadoに送られ、そこからCeleryパブリッシャーを介してRabbitMQに送られ、そこからCeleryコンシューマーを介して処理され、MongoDBに送信されました。 1秒以内（SLA）にユーザーの推奨リクエストに応答する必要があります。

問題

最初は、システムはうまく機能しました。イベントのストリームを取得して処理し、推奨事項を計算しました。 しかし、すぐに次の問題が明らかになりました。



まず、技術的な問題。 人気のあるビデオに対する推奨の再計算は、システム全体を「レイアウト」しました。 Mongoからの膨大な数のイベントがすべてのRAMを詰まらせ、これらすべてのイベントを処理するとすべてのコアが詰まりました。 そのような場合、すでに計算された推奨事項を返還したり、SLAに違反して新しい推奨事項を再集計したりすることはできません。



第二に、ビジネス上の問題。 フィボナッチシリーズに属するi回ごとに推奨事項を数えるだけで、バイラルビデオをすぐに追跡できませんでした。 ようやく彼に着いたとき、イベントの数が増え続けながら、完全な再集計を行うたびに、時間とリソースがかかりすぎました。



第三に、暗黙的なフィードバック問題と呼ばれるアルゴリズムの問​​題。 推奨システムは、評価または評価に基づいています。 ビデオビューのみを収集しました。つまり、実際には、評価システムはバイナリでした。1-ビデオを視聴し、0-視聴しませんでした。

解決策

解決策として、完全なリアルタイムへの移行が提案されましたが、そのためにはリソースの観点から安価なアルゴリズムと、同時に効率的なアルゴリズムが必要でした。 主なインスピレーションは、 Tencentが発行した記事でした 。 新しい推奨エンジンの基礎として、彼らが説明したアルゴリズムを説明しました。

アルゴリズム：パーソナライズされた推奨事項

新しいアルゴリズムは、古典的なアイテムベースのCFであり、角度余弦が近接性の尺度として使用されました。











数式を注意深く見ると、分子の共同評価の合計と分母の評価の2つの平方和の3つの合計で構成されていることがわかります。 それらをより単純な概念に置き換えると、次のものが得られます。



























では、新しいイベント（新しい評価）を受け取ったときに推奨事項を再計算するとどうなるかを考えてみましょう。 そう！ 受け取った値だけ金額を増やします。 したがって、近接度の新しい尺度は次のように表すことができます。











これは非常に重要な結論につながります。 毎回すべての金額を再集計する必要はありません。 それらを保存し、新しい値ごとに増やすだけで十分です。 この原則は増分更新と呼ばれます。 彼のおかげで、推奨事項を再計算するときの負荷が大幅に削減され、リアルタイム処理でそのようなアルゴリズムを使用することが可能になります。



同じ記事では、暗黙的なフィードバックの問題に対する解決策が提案されました。ビデオを見ているときの特定のユーザーアクションを評価することです。 たとえば、視聴の開始、中間または終了までの視聴、一時停止、巻き戻しなどを個別に評価します。 重みの設定方法は次のとおりです。

ACTION_WEIGHTS = { "thirdQuartile": 0.75, "complete": 1.0, "firstQuartile": 0.25, "exitFullscreen": 0.1, "fullscreen": 0.1, "midpoint": 0.5, "resume": 0.2, "rewind": 0.2, "pause": 0.05, "start": 0.2 }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

1つのビデオで複数のユーザーアクションを行うと、最大評価が取得されます。

アルゴリズム：人気の動画

また、今では人気のある（トレンドの）ビデオを考慮しています。 各ビデオにはカウンタがあり、それに関連付けられたイベントを受信すると、1ずつ増加します。 最も人気のある動画は上位20位です。 新しいイベントが到着するたびに、ビデオがトップ20にあるかどうかを確認します。そうでない場合は、トップ20のすべてのビデオのカウンターが1ずつ減少します。したがって、かなり動的なリストが取得され、ウイルスビデオを簡単に追跡できます。 地域ごとにこのようなリストを維持しています。 つまり、さまざまな地域のユーザーにさまざまな人気のある動画をお勧めします。



ユーザーに推奨事項を要求する場合、問題は個人的な推奨事項の85％と人気のあるビデオの15％で構成されます。

テクノロジー

アルゴリズムを決定しました。次に、それを実装できるテクノロジーを選択する必要があります。 ストリーム処理で見つかったすべてのケースを要約すると、使用されるテクノロジーのスタックは次のようになります。







つまり、すべてのイベントをメッセージブローカーに送信し、ストリーム処理エンジンでそれらを並行して処理し、結果をキー値データベースに追加します。そこからリクエストの受信時に読み取ります。 各アイテムに特定のオプションを選択することは残っています。



すでにRabbitMQがあり、非常に満足しています。 ただし、ほとんどのストリーミングエンジンはボックスからのKafkaをサポートしています。 特に最近、Kafkaが大きな勢いを獲得し、それ自体が十分に証明され、一般的にはかなり普遍的で信頼できるツールであると見なされているため、移行にまったく反対しませんでした。



ストリーミングエンジンとして、3つのオプションすべてを検討しました。 ただし、すぐに1つの制限が導入されました。すべてのコードはPythonでのみ記述する必要があります。 私たちのチームのすべてのメンバーは彼をよく知っていますし、Pythonには必要なすべてのライブラリがあり、そこになかったライブラリは自分で書きました。 したがって、SamzaはJavaのみをサポートするため、すぐに削除されました。 最初はStormを試してみたかったのですが、ドキュメントを掘り下げた後、コードの一部（トポロジ）はJavaで記述され、処理ノード（ボルト）のコードのみがPythonで記述できることがわかりました。 後に、 streamparseと呼ばれるStormの美しいPythonラッパーを発見しました。 PythonでStormを完全に必要としている人は、お勧めします！ その結果、選択はSparkに落ち着きました。 まず、私たちはすでに彼を知っていました。 次に、Pythonを完全にサポートしました。 そして第三に、彼らが現在どのくらい急速に発展しているのか、本当に印象的です。



データベースの選択はすぐに行われました。 それぞれに経験があるにもかかわらず、お気に入りがあります。 これはエアロスパイクです。 Aerospikeは、SSDドライブを備えたサーバーでその可能性を最大限に引き出します。 Linuxファイルシステムをバイパスし、SSDに個別のブロックで直接書き込みます。 このため、1つのノードで100万TPSのパフォーマンスが達成され、要求の99％の応答時間は1ミリ秒未満です。 私たちはそれを複数回使用しており、料理はかなり上手です（ただし、料理することはあまりありませんが）。



その結果、スタックは次のようになります。







ストームとスパークの選択についてのもう少しの言葉、それは突然それが誰かのために役立つでしょう。



Sparkストリーミングはリアルタイムエンジンではなく、ミニバッチ処理を使用するため、かなりリアルタイムに近いエンジンです。 つまり、データストリームは、指定された秒数のサイズのミニバッチに分割され、並列処理されます。 Sparkの概念に精通している人にとって、ミニバッチはRDDに他なりません。 Stormはリアルタイムです。 着信イベントはすぐに処理されます（ただし、StormはTridentを使用してミニバッチ処理するように構成することもできます）。 要約すると、Stormは待ち時間が短く、Sparkはスループットが高くなります。 レイテンシに厳しい制限がある場合は、Stormを選択する必要があります。 たとえば、セキュリティアプリケーション用。 そうでなければ、Sparkで十分です。 運用環境で1か月以上使用しており、クラスターを適切に調整して時計のように機能します。 Sparkはよりフォールトトレラントであるという意見もあります。 ただし、私たちの意見では、補足する必要があります; Sparkは、すぐに使用できるフォールトトレラントです。 十分な欲望とスキルで、Stormはそのようなレベルに持っていくことができます。

結果

そして最も重要なことです。 それはすべて理にかなっていますか？ 持っていた 推奨事項の品質から始めましょう。 オンライン品質指標を使用します。これは、ユーザーが実際に視聴したビデオが、当社が推奨するリストに含まれているかどうか、およびその場所を測定します。 配布は次のとおりです。



個人的な推奨事項のみ（item-item CF）：







人気のあるアイテムのみ（トレンド）：







ハイブリッドの推奨事項（85％のアイテム-アイテムCFおよび15％のトレンド）：







これは、メトリックが最大であるため、本番環境で使用する最後のオプションです。 つまり、ユーザーが視聴した動画の約10％が、そもそもお勧めです！ 約36％がトップ100に入っています。これらの数値から何もわからない場合は、人気のある動画の分布を確認できます。 人気のある動画のみをユーザーに推奨する場合、ユーザーが視聴したすべての動画の0.1％だけを推測します。 これは、ビデオを推測する確率が1 /（126万）であるランダム分布と比較した場合にも優れています。



明確にするために、以下にメトリックのグラフを示します。







14:00から15:30までの12.02の部分。最も予期しない（赤）は、トレンド項目のみで推奨事項をテストするだけです。 その前-アイテム-アイテムCF、そしてその後-ハイブリッドモデル。 ご覧のとおり、「リアルタイム」のおかげで、当社の推奨システムは新しい構成にすばやく適応します。 ちなみに、システムを最初から起動すると、上記のインジケータに30分間表示されます。



MongoDBからAerospikeへの移行のおかげで、推奨事項の要求に対する平均応答時間が短縮されました。 以下は、古いエンジンから新しいエンジンに切り替える瞬間です。







ご覧のとおり、応答時間は約5倍になりました。 これは、SLAについてもう心配しないので、私たちにとってもRutube自体にとっても快適です。



鉄に関しては、3台のマシンに新しいクラスター（Cloudera）をデプロイしました。 今のところ、これで十分です。 システム全体がフォールトトレラントなので、稼働率は99.9％です。 これに加えて、スケーラビリティがあります。クラスターの負荷を大幅に増加させながら、いつでも新しいマシンを「固定」できます。

次は何ですか

ここで2つのポイントに興味があります。イベントの重み（視聴の開始、途中までの視聴、一時停止など）と、推奨の発行における人気のある動画のシェアです。 これらの指標と他の指標の両方は、専門家の方法、つまり「目で」によって設定されました:)それにもかかわらず、両方のケースで最も効果的な指標を決定したいと思います。 ほとんどの場合、 Metric Optimization Engine（MOE）を使用します。 これは、A / Bテストに基づいてシステムパラメータを決定するためのフレームワークです。 興味深い結果が得られたら、すぐに書いてください。



ケースとストリーム処理全般について質問がある場合は、コメントでお尋ねください。全員に回答するよう努めます。