ビッグデータを操作するための2つのYandexライブラリに関する講義

数週間前、Yandexで、Pythonを使用したビッグデータの分析に特化したPyData会議が開催されました。 この会議を含め、Yandex分析開発グループの責任者であるVasily Agapitovが話しました。 彼は、MapReduceで計算を記述して実行するためと、ログから情報を抽出するための2つのライブラリについて話しました。





カットの下-デコードとスライドの一部。

---

私の名前はVasily Agapitovです。データマイニングチームを代表しています。



Yandexでは、ビッグデータ、特にMapReduceクラスターにあるデータを使用して計算を実行します。 これらは主にサービスとアプリケーションの匿名ログです。 さらに、ビッグデータ処理ツールを他のチームに提供しています。 私たちの主な消費者は開発アナリストのチームです。 簡単にするために、私はそれらをアナリストと呼びます。



2つのツール、ライブラリ、それらの外観の歴史、およびHadoopの世界が外観にどのように影響したかについてお話します。







用語のアイデアを同期させましょう。 データソース。 ログについてのみ説明します。たとえば、ある種のサービスアクセスログを見てみましょう。







このログには水平エントリがあり、各エントリにはフィールドがあります。vhostはホスト識別子、yandexuidは訪問者識別子、iso_eventtimeはリクエストの日付と時刻、requestはリクエスト自体およびその他の多くのフィールドです。



一部のフィールドのデータはすでに計算に使用できます。 他のフィールドから、データを最初に抽出して正規化する必要があります。 たとえば、リクエストフィールドにはリクエストパラメータが含まれます。 各サービスには独自のパラメーターがあります。 検索に最もよく使用されるパラメーターはテキストパラメーターです。 リクエストフィールドからそれを抽出した後、それを非常に大きくするか、何らかの奇妙なエンコーディングを持つ可能性があるため、正規化する必要があります。







次に、MapReduceクラスターでの計算の観点から計算を検討します。 ご存知のように、MapReduceはサンドイッチ作成テクノロジーです。 いいえ、これはビッグデータ処理技術です。 慣れていない場合、現在のレポートでは、MapとReduceの2つの操作を使用したデータ処理が含まれていることを知っておく必要があります。



アナリストのタスクは、ビジネスロジックを考慮して、ログのイベントに関する計算を作成することです。 ライブラリを使用せずにMapReduceクラスターでタスクを解決する場合、アナリストはどのような困難に直面するでしょうか？ まず、MapReduceの操作に基づいてビジネスロジックを実装する必要があります。 このアプローチは、この計算のビジネスロジックに関連しない計算コードに追加され、読みやすさとサポートを著しく損ないます。 第二に、最初にログからデータを抽出して正規化する必要があります-たとえば、リクエストフィールドからのテキストパラメータ。



最初の問題を解決するには？ 明らかに、クラスターへのユーザーアクセスとクラスターとの対話を簡素化する何らかの種類のライブラリが必要です。



Hadoopの世界では、そのようなライブラリには、Pig、Hive、Cascadingなどが含まれます。



YandexはYTと呼ばれるMapReduceの独自の実装を使用します。この利点は、Habréの記事で読むことができ、データ処理のための基本的なMapReduce操作を提供します。 しかし、残念ながら、YTにはHadoopの世界のライブラリの類似物がありませんでした。 修正する必要がありました。



最初に、この問題が発生したときに、計算ごとにマップステージを個別に説明し、ステージを個別に削減し、クラスターで計算を開始するこれらのステージ間の関係を個別に説明しました。







さらに、それぞれに独自のランチャーがありました。 そのような動物園を維持することは非常に高価です。 私たちの解決策は、クラスターで計算を記述して実行するためのライブラリであるナイルライブラリでした。 Pythonを作成したとき、主にPythonがローカルデータ処理に分析を使用するため、Hadoopの世界からCascadingのアイデアを取り入れてPython言語で実装しました。クラスター上の計算とローカルデータ処理に1つの言語を使用すると非常に便利です。







Cascadingを知っている場合、Nileでデータを処理するプロセスもおなじみのように思えます。 クラスター上のテーブルからストリームを作成、変更、グループ化、たとえば、いくつかのアグリゲートを検討し、ストリームを複数のフローに分割し、複数のフローを1つのストリームに結合し、他のアクションを実行した後、結果のストリームを必要なデータとともにテーブルに保存しますクラスター。



ナイルにはどのようなストリーム変更操作がありますか？ それらの多くがあります、最も使用されるものはここに提示されます。 必要なフィールドのリストを取得するプロジェクト。 必要なレコードのみを残して、フィルターからフィルターへ。 Groupby + Aggregateを使用して、特定のフィールドセットでストリームをグループ化し、いくつかの集計をカウントします。 一意、ランダム、および固有の、ランダム、および指定されたレコード番号を作成します。 指定されたフィールドのセットが等しいことにより、2つのストリームを結合する結合。 分割して、いくつかのルールに従ってストリームをいくつかのストリームに分割し、それぞれをさらに個別に処理します。 並べ替えて並べ替えます。 テーブルをクラスターに配置します。



MapおよびReduce操作も使用できますが、本当にカスタムで複雑な操作を行う必要がある場合はほとんど必要ありません。







ナイルの初期化を見てみましょう。 彼女はとてもシンプルです。 インポート後、クラスターとジョブの2つのオブジェクトを作成します。 クラスターは、どのクラスターで実行するか、および他の例示的な環境を示すために必要です。 ジョブ-ストリームを変更するプロセスを説明します。







ストリームを作成する方法は？ ストリームは、クラスター上のテーブルから、または既存のスレッドからの2つの方法で作成できます。 最初の2つの例は、クラスター上のテーブルからストリームを作成する方法を示しています。 クラスター上のテーブルへのパスが引数として渡されました。 最後の例は、2つの既存のストリームをマージしてストリームを作成する方法を示しています。



ナイルでのタスクの実装例を見てみましょう。







アクセスログにより、yandex.com.trホストの訪問者数を計算する必要があります。 アクセスログの様子を思い出してください。 表示される多くのフィールドのうち、vhostフィールドは、yandex.com.trホストに関連するエントリのみをフィルタリングして残し、yandexuidは訪問者数を計算します。



このタスクのNileコード自体は次の形式になります。







ここで、クラスターのテーブルからストリームを作成し、vhostおよびyandexuidフィールドを取得します。 vhostフィールドの値がyandex.com.trに等しいレコードのみを残し、yandexuidフィールドの一意の値の数を考慮した後、クラスターのテーブルにストリームを保存します。 Job.run（）は計算を開始します。







クラスターで計算を開始する前に、Ni​​leは計算を一連のMapReduce操作に変換します。 左側はナイルに関するストリーム変換グラフ、右側はMapReduce操作に関するストリーム変換グラフです。 さらに、Nileは計算を自動的に最適化します。つまり、複数のMap操作が連続している場合、Nileはそれらをクラスターで実行される1つのMap操作に接着できます。 何が起こったのかがわかります。 これは非常に単純なタスクであり、そのためのコードはプログラミング言語で見るのが比較的簡単です。



より複雑なことを考えるために、アナリストが直面する2番目の問題を思い出してみましょう。ログからのデータを抽出し、正規化する必要があります。



この問題を解決することはどのように決定されますか？ 通常、 ETLはこの問題を解決するために使用されます。 誰がそれを知っていますか？ 彼は約30パーセントを知っています。 誰が知らないかわからないが、忘れてしまった：ETLは生の生データがあると仮定し、それらから必要なレコード、フィールドなどを抽出し、ビジネスロジックを考慮してそれらを変更し、リポジトリにロードします。 将来的には、ストレージからのデータ、つまり正規化されたデータに基づいてすべての計算を実行します。



別のパスを選択しました。 生データを保存し、計算を実行します。データを抽出して正規化するプロセスは、各計算で発生します。



なぜこの方法を選んだのですか？ フィールドの抽出および正規化のプロセスで外部ライブラリが使用され、このライブラリにバグがあったとします。 バグを修正した後、過去の計算を再計算する必要があります。 このアプローチでは、過去の計算を実行し、正しい結果を取得するだけです。 ETLの場合-このライブラリがプロセスで使用された場合-最初にデータを再度抽出し、このライブラリで処理し、リポジトリに配置してから、計算を実行する必要があります。



生データと正規化データの両方を保存できる場合に適したオプションです。 残念ながら、大量のデータがあるため、これは不可能です。



まず、計算ごとに、データの抽出と正規化を個別に実行しました。 次に、同じログに対して、ほぼ同じ方法で同様のフィールドを取得することが多いことに気付き、これらのルールを1つのQB2ライブラリに結合しました。 ルール自体はエクストラクターと呼ばれ、ログに関するこのようなルールのセットは解析されたツリーでした。







そのため、QB2ライブラリは生ログへの抽象インターフェイスを提供し、ツリーの解析について認識しています。







解析ツリーを理解しましょう。 これは、Yandexモバイルアプリケーションのログの解析ツリーです。 よく見ないでください、これはグローバルマップです。



一番上はログエントリです。 他のすべての色付きの長方形は仮想フィールドです。 接続は抽出機能です。 したがって、QB2ライブラリを持たないユーザーは、ログフィールドにのみアクセスできます。 QB2ライブラリを使用するユーザーは、ログフィールドと、このログに対してQB2ライブラリが提供するフィールドの両方を使用できます。



このライブラリの利点を具体的な例で見てみましょう。 問題を検討してください。 10321に等しいAPI_keyフィールドの値で識別されるYandexモバイルアプリケーションがあるとします。フィールド、特にJSONオブジェクトの形式で辞書を含むevent_valueフィールドにログを書き込むことがわかります。 この辞書のステージキーの意味に興味があります。



イベントの日付ごとに分解して、各ステージの訪問者数を計算する必要があります。 QB2ライブラリはこれにどのようなフィールドを提供しますか？ まず、API_key、アプリケーション識別子。 次に、device_id、ユーザーID。 最後に、イベントの日付であるevent_dateは、ログエントリから直接取得されるのではなく、かなり多数の変換によって取得されます。 QB2ライブラリを使用しないと、このフィールドが必要な各計算で手動変換を実行する必要があります。 同意します、これは不便です。



さらに、event_valueフィールドとステージ値が必要です。 それらはツリーでは解析されません。これは論理的です。なぜなら、それらは1つの特定のアプリケーションに対してのみ書かれているからです。



エクストラクターを使用して、解析ツリーを次のフォームに追加する必要があります。 このタスクの初期化はどのように変わりますか？ さらに、エクストラクターとフィルターをインポートします。 event_valueおよびステージ値を取得するには、エクストラクターが必要です。 フィルター-レコードをフィルターし、必要なものだけを残します。







前の例ではナイルからのフィルタリングを使用していましたが、なぜこの例ではQB2ライブラリからのフィルタリングを使用したのでしょうか？ QB2は、Nileと同様に、計算の最適化を試みます。つまり、フィルタリングで使用されるフィールドの値をできるだけ早く取得しようとします。 他の仮想フィールドの値よりも早い。



なぜこれが行われるのですか？ レコードがフィルタリング条件を通過しない場合、残りの仮想フィールドの値を取得しないようにします。 したがって、コンピューティングリソースを大幅に節約し、クラスターでの計算を高速化します。







計算コード自体は次の形式になります。 ここでは、クラスターのテーブルからストリームを作成し、QB2演算子でストリームを変更します。これは、解析ツリーの名前（この場合はログの名前と一致し、フィールドとフィルターのセットと一致します）で初期化します。



フィールドには、API_key、device_id、event_dateフィールドを名前でリストします。これは、QB2ライブラリがこれらのフィールドを取得する方法をすでに知っているためです。



event_valueフィールドを抽出するには、標準のjson_log_field抽出プログラムを使用します。 彼は何をしていますか？ 渡された名前で、対応するログフィールドから値を受け取り、JSONオブジェクトとしてロードします。 このロードされた値をevent_valueフィールドに保存します。



ステージ値を取得するには、別の標準エクストラクターdictItemを使用します。 彼に渡されたキー名とフィールド名を使用して、このキーのこのフィールドから対応する値を抽出します。



フィルタリングについて。 device_idフィールド値が定義され、アプリケーションに関連するレコード、つまりapi_keyフィールド値が10321であるレコードにのみ関心があります。QB2演算子を適用した後、ストリームには次のフィールドがあります：api_key、device_id、event_date、event_value、stage。



結果のストリームを次のように変更します。 event_dateフィールドとstageフィールドのペアでグループ化し、一意のdevice_id値の数を計算します。 この値をusersフィールドに入れ、その後、結果のストリームをクラスターに保存します。 Job.run（）は計算を開始します。







計算が完了すると、クラスターには次の形式のテーブルがあります。event_dateフィールドとstageフィールドの各ペアについて、一意のユーザー値の数はusersフィールドにあります。 したがって、QB2とNileライブラリの統合により、私が表明した両方の問題が解決されます。 ご清聴ありがとうございました。