ビデオサービスであれ、テープであれ、ストーリーであれ記事であれ、どんな製品でも開発して、ユーザーの条件付きの「幸福」を測定できるようにしたい。 私たちが変化を良くするか悪くするかを理解し、直観や私たち自身の感情ではなく、あなたが信じることができる指標と数値に依存して、製品開発の方向を調整します。



この記事では、非常に高いパフォーマンスの分析クエリを取得しながら、月間9,700万人のオーディエンスを抱えるサービスで製品統計と分析を開始した方法について説明します。 ClickHouse、使用されるエンジン、クエリの機能について説明します。 データ集約のアプローチを説明します。これにより、複雑なメトリックを一瞬で取得し、データの変換とテストについて話すことができます。



現在、1日あたり約60億の製品イベントがあり、近い将来、200〜250億に達する予定です。 そして、そのような速いペースではなく、私たちが関心のあるすべての食物イベントを説明する年末までに、400〜500億に上昇します。



セット内の1行。 経過時間：0.287秒 処理された598.5億行、59.85 GB（208.16億行/秒、208.16 GB /秒）



カットの下の詳細。

まえがき

分析ツールは以前はVKontakteでした。 一意のユーザーが考慮されたため、スライスごとにイベントスケジュールを作成し、サービスの深層に分類することができました。 しかし、彼らは事前に固定スライスについて、集計データについて、ユニークなものについてはHLLについて、「いくら？」よりもやや複雑な質問に素早く答えることができない硬さについて説明しました。



もちろん、Hadoopがあり、そうであり、またそうであり、サービスの使用に関する多くのログも書き込まれ、書き込まれ、書き込まれます。 残念ながら、hdfsは一部のチームが独自のタスクを実装するためにのみ使用されていました。 さらに悲しいことに、hdfsは迅速な分析クエリに関するものではありません。多くの分野への質問があり、その答えはコードにあり、誰もがアクセスできるドキュメントにはありませんでした。



私たちは、このように生きることはもはや不可能であるという結論に達しました。 各チームにはデータが必要であり、それらに対するクエリは高速であり、データ自体は正確で有用なパラメーターが豊富である必要があります。



そのため、新しい統計/分析システムの明確な要件を策定しました。

• 分析クエリは高速でなければなりません。
• データは非常に正確です。理想的には、これらはサービスとの生のユーザーインタラクションイベントです。
• イベントの構造を説明し、理解し、アクセス可能にする必要があります。
• 信頼できるデータストレージ、1回限りの配信保証。
• ユニーク、オーディエンス（毎日、毎週、毎月）、保持メトリック、ユーザーがサービスに費やした時間、ユニークセットに対する定量化されたアクション、およびスライスのセットごとのその他のメトリックをカウントできます。
• テスト、データ変換、視覚化が進行中です。

キッチンで

経験から、2つのデータベースが必要であることが示唆されました。1つはデータを集約して強化する低速のデータベース、もう1つはこのデータを操作してその上にグラフを作成できる高速のデータベースです。 これは最も一般的なアプローチの1つです。たとえば、hdfsのような遅いベースでは、特定の期間のスライスごとのユニークな投影とイベント数に基づいて、異なる投影が作成されます。



暖かい9月の日に、カザン大聖堂を見下ろすキッチンでお茶を飲みながら会話をしながら、ClickHouseをクイックベースとして試してみるというアイデアがありました。当時は、それをすでに技術ログの保存に使用していました。 主に速度と信頼性に関連する多くの疑問がありました。宣言されたパフォーマンステストは非現実的であり、新しいデータベースリリースは定期的に既存の機能を破壊しました。 したがって、提案は簡単でした-試してみてください。

最初のサンプル

この構成で2台のマシンのクラスターをデプロイしました。

2xE5-2620 v4（合計32コア）、256Gラム、28Tプレース（raid10 with ext4）。



最初はレイアウトに近いものでしたが、それから遠くに切り替えました。 ClickHouseにはさまざまなテーブルエンジンがありますが、主なものはMergeTreeファミリのものです。 おおよそ次の設定でReplicatedReplacingMergeTreeを選択しました。

PARTITION BY dt ORDER BY (toStartOfHour(time), cityHash64(user_id), event_microsec, event_id) SAMPLE BY cityHash64(user_id) SETTINGS index_granularity = 8192;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

複製 -テーブルが複製されることを意味します。これにより、信頼性の要件の1つが解決されます。



置換 -テーブルは主キーによる重複排除をサポートします。デフォルトでは、主キーはソートキーと一致するため、ORDER BYセクションは主キーとは何かを示すだけです。



SAMPLE BY-サンプリングを試してみたかったのです。sampleは一様に疑似ランダムなサンプルを返します。



index_granularity = 8192は、インデックスのセリフ間のデータ行の魔法の数です（はい、まばらです）。これはデフォルトで使用されます。 変更しませんでした。



パーティション分割は日単位で行われました（ただし、デフォルトでは月単位）。 多くのデータ要求は日中にあると想定されていました。たとえば、特定の日のビデオビューの分グラフを作成するためです。



次に、技術ログを取得して、テーブルに約10億行を入力しました。 優れた圧縮、列タイプInt *によるグループ化、一意の値のカウント-すべてが非常に高速に動作しました！



速度について言えば、1つの要求が500ミリ秒より長く持続せず、それらのほとんどが50〜100ミリ秒に収まることを意味します。 そして、これは2台のマシン上にあります。実際、計算に関与したのは1台だけです。



すべてを見て、UInt8列の代わりに国のIDがあり、Int8列がユーザーの年齢などのデータに置き換えられることを想像しました。 そして彼らは、すべてが正しく行われていれば、ClickHouseが私たちに完全に適していることに気付きました。

強力なデータ型付け

ClickHouseの利点は、正しいデータスキーマが形成されたときに正確に始まります。 例：プラットフォームString-悪い、プラットフォームInt8 +辞書-良い、LowCardinality（文字列）-便利で良い（LowCardinalityについては後で説明します）。



特別なジェネレータークラスをphpで作成しました。これは、リクエストに応じて、ClickHouseのテーブルに基づくイベントのラッパークラスと、ロギングの単一のエントリポイントを作成します。 判明したスキームの例を説明します。

1. アナリスト/データエンジニア/開発者がドキュメントについて説明します：どのフィールド、可能な値、イベントをログに記録する必要があるか。
2. 前の段落のデータ構造に従って、ClickHouseにテーブルが作成されます。
3. テーブルに基づくイベントのラップクラスが生成されます。
4. 製品チームは、このクラスのオブジェクトのフィールドへの入力、送信を実装します。

PHPのレベルでスキームを変更し、記録されたデータのタイプを変更するには、最初にClickHouseのテーブルを変更する必要があります。 そしてこれは、チームとの調整、ドキュメントの変更、イベントの説明なしでは実行できません。



イベントごとに、ClickHouseとhadoopに送信されるイベントの割合をそれぞれ制御する2つの設定を設定できます。 設定は、何かがうまくいかない場合にロギングを削減する機能を備えた段階的なローリングのために主に必要です。 hadoopの前に、データはKafkaを使用して標準的な方法で配信されます。 また、ClickHouseでは、永続モードのKittenHouseを使用して、少なくとも1つのイベント配信を保証するスキームを使用します。



イベントは、user_idからのハッシュをクラスター内のシャードの数で割った残りに基づいて、目的のシャードのバッファーテーブルに配信されます。 次に、バッファテーブルはデータをローカルのReplicatedReplacingMergeTreeにフラッシュします。 また、ローカルテーブルの上に、分散エンジンで分散テーブルがプルされます。これにより、すべてのシャードのデータにアクセスできます。

非正規化

ClickHouseは、カラムナーDBMSです。 それは、通常の形式に関するものではありません。つまり、参加するよりも、イベントですべての情報を持っている方が良いということです。 Joinもありますが、適切なテーブルがメモリに収まらない場合、痛みが始まります。 そのため、私たちは強い意思を決定しました。関心のあるすべての情報はイベント自体に保存する必要があります。 たとえば、性別、ユーザーの年齢、国、都市、誕生日-これらはすべて、オーディエンス分析に役立つ公開情報であり、インタラクションオブジェクトに関するすべての有用な情報です。 たとえば、ビデオについて話している場合は、video_id、video_owner_id、ビデオのアップロード日、長さ、イベント時の品質、最大品質などです。



合計で、各テーブルには50〜200列がありますが、すべてのテーブルにはサービスフィールドがあります。 たとえば、エラーログはerror_logです。実際、タイプの範囲外のエラーを呼び出します。 奇妙な値が年齢のフィールドの型のサイズを超える場合。

タイプLowCardinality（T）

ClickHouseには、外部辞書を使用する機能があります。 それらはメモリに保存され、定期的に更新され、古典的な参考書など、さまざまなシナリオで効果的に使用できます。 たとえば、オペレーティングシステムをログに記録する場合、2つの選択肢があります。文字列または数字+ディレクトリです。 もちろん、大量のデータに対して、および高性能の分析クエリの場合、数値を記述し、必要に応じて辞書から文字列表現を取得することは論理的です。

 dictGetString('os', 'os_name', toUInt64(os_id))
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

しかし、もっと便利な方法があります-タイプLowCardinality（String）を使用して、辞書を自動的に作成します。 値のセットのカーディナリティが低いという条件下でのLowCardinalityのパフォーマンスは、Stringの場合よりも大幅に高くなります。



たとえば、イベントタイプ「play」、「pause」、「rewind」にLowCardinality（String）を使用します。 または、プラットフォームの場合：「web」、「android」、「iphone」：

 SELECT vk_platform, count() FROM t WHERE dt = yesterday() GROUP BY vk_platform Elapsed: 0.145 sec. Processed 1.98 billion rows, 5.96 GB (13.65 billion rows/s., 41.04 GB/s.)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

この機能はまだ実験的であるため、使用するには以下を実行する必要があります。

 SET allow_experimental_low_cardinality_type = 1;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

しかし、しばらくすると、彼女はもはや設定下に置かれないという感じがあります。

VKデータ集約

多くの列があり、多くのイベントがあるので、自然な欲求は「古い」パーティションをカットすることですが、最初にユニットを組み立てることです。 場合によっては、生のイベント（1か月または1年前）を分析する必要があるため、hdfsのデータをカットしません。アナリストは、希望する寄木細工にいつでも連絡できます。



原則として、時間間隔で集計する場合、単位時間あたりの行数はカット電力の積に等しいという事実に常に基づいています。 これは制限を課します：国は、「ロシア」、「アジア」、「ヨーロッパ」、「その他の世界」、および年齢などのグループで収集を開始します。

dt、user_idによる集計

しかし、リアクティブClickHouseがあります！ 日付で5,000〜1億行に加速できますか？

簡単なテストにより、ユーザーをマシンに残すという簡単なアイデアが生まれたことがわかりました。 つまり-sparkを使用した「日付、スライス」ではなく、ClickHouseを使用した「日付、ユーザー」で集計し、データの「転置」を行います。



このアプローチでは、ユーザーを集計データに保存します。つまり、ユーザーインジケーター、保持率、頻度の指標をカウントできます。 複数のサービスの共通オーディエンスをVKontakteオーディエンス全体までカウントして、ユニットを接続できます。 これはすべて、条件付きで同時にテーブルに存在する任意のスライスによって実行できます。



例で説明します。







集約後（右側にさらに多くの列があります）：







この場合、集約は（dt、user_id）によって正確に行われます。 このような集計でユーザー情報を持つフィールドの場合、関数any、anyHeavyを使用できます（頻繁に発生する値を選択します）。 たとえば、anyHeavy（プラットフォーム）を集約に収集して、ユーザーがビデオイベントの大部分で使用しているプラ​​ットフォームを知ることができます。 必要に応じて、groupUniqArray（プラットフォーム）を使用して、ユーザーがイベントを発生させたすべてのプラットフォームの配列を格納できます。 これで十分でない場合は、プラットフォームとストアに個別の列を作成できます。たとえば、特定のプラットフォームから半分にスクリーニングされた一意のビデオの数を作成できます。

 uniqCombinedIf(cityHash64(video_owner_id, video_id), (platform = 'android') AND (event = '50p')) as uniq_videos_50p_android
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このアプローチでは、各行が一意のユーザーであり、各列にユーザーまたはサービスとの相互作用に関する情報が含まれる、かなり広い集計が取得されます。



サービスのDAUを計算するには、そのユニット上でそのようなリクエストを実行するだけで十分であることがわかります。

 SELECT dt, count() as DAU FROM agg GROUP BY dt Elapsed: 0.078 sec.
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

または、ユーザーが1週間サービスを利用していた日数を計算します。

 SELECT days_in_service, count() AS uniques FROM ( SELECT uniqUpTo(7)(dt) AS days_in_service FROM agg2 WHERE dt > (yesterday() - 7) GROUP BY user_id ) GROUP BY days_in_service ORDER BY days_in_service ASC 7 rows in set. Elapsed: 2.922 sec.
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

精度をほとんど損なうことなく、サンプリングによって加速できます。

 SELECT days_in_service, 10 * count() AS uniques FROM ( SELECT uniqUpTo(7)(dt) AS days_in_service FROM agg2 SAMPLE 1 / 10 WHERE dt > (yesterday() - 7) GROUP BY user_id ) GROUP BY days_in_service ORDER BY days_in_service ASC 7 rows in set. Elapsed: 0.454 sec.
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

サンプリングはイベントの割合ではなく、ユーザーの割合によって行われることに注意してください。その結果、サンプリングは非常に強力なツールになります。



または、1/100サンプリングで4週間同じ-約1％精度の低い結果が得られます。

 SELECT days_in_service, 100 * count() AS uniques FROM ( SELECT uniqUpTo(7)(dt) AS days_in_service FROM agg2 SAMPLE 1 / 100 WHERE dt > (yesterday() - 28) GROUP BY user_id ) GROUP BY days_in_service ORDER BY days_in_service ASC 28 rows in set. Elapsed: 0.287 sec.
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

一方、集計

（dt、user_id）で集計する場合、ユーザーが失われることはありません。サービスとのやり取りに関する情報を見逃すことはありませんが、もちろん、特定のやり取りオブジェクトに関するメトリックは失われます。 しかし、あなたもこれを失うことはできません-ユニットを構築してみましょう

（dt、video_owner_id、video_id）、同じアイデアを堅持。 動画に関する情報は可能な限り保持し、動画とユーザーとのやり取りに関するデータを見逃すことはありません。特定のユーザーに関する情報は完全に見逃します。

 SELECT starts FROM agg3 WHERE (dt = yesterday()) AND (video_id = ...) AND (video_owner_id = ...) 1 rows in set. Elapsed: 0.030 sec
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

または、昨日のトップ10のビデオビュー：

 SELECT video_id, video_owner_id, watches FROM video_agg_video_d1 WHERE dt = yesterday() ORDER BY watches DESC LIMIT 10 10 rows in set. Elapsed: 0.035 sec.
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

その結果、次の形式の集約のスキームがあります。

• 製品内の「日付、ユーザー」による集計。
• 製品内の「日付、相互作用のオブジェクト」による集計。
• 他の予測が発生する場合があります。

アズカバンとTeamCity

最後に、インフラストラクチャについて少し説明します。 集計コレクションは夜間に開始され、各生データテーブルのOPTIMIZEから始まり、ReplicatedReplacingMergeTreeで異常なデータマージをトリガーします。 操作は十分に長く続くことができますが、重複が発生した場合は削除する必要があります。 これまでのところ、重複に遭遇したことはありませんが、将来それらが表示されないという保証はありません。



次のステップは、集計の作成です。 これらは、以下が発生するbashスクリプトです。

• まず、シャードの数とシャードからいくつかのホストを取得します。
  
  
  
  

   SELECT shard_num, any(host_name) AS host FROM system.clusters GROUP BY shard_num
        
        
  
          
          
          
        
  
      
          
          
          
        
        
  
          
          
          
        
  
      
       

• 次に、スクリプトは各シャード（clickhouse-client -h $ host）に対してフォームのリクエスト（ユーザーによる集計用）を順番に実行します。
  
  
  
  

   INSERT INTO ... SELECT ... FROM ... SAMPLE 1/$shards_count OFFSET 1/$shard_num
        
        
  
          
          
          
        
  
      
          
          
          
        
        
  
          
          
          
        
  
      
       

これは完全に最適なものではなく、ホスト間で多くのネットワーク相互作用を引き起こす可能性があります。 ただし、新しいシャードを追加するときは、すべてがそのまま使用でき、ユニットのデータの局所性が維持されるため、あまり心配しないことにしました。



タスクスケジューラとしてAzkabanがあります。 これは非常に便利なツールとは言いませんが、やや複雑なパイプラインの構築や、1つのスクリプトが他のいくつかのスクリプトの完了を待つ必要がある場合など、そのタスクに完全に対応しています。



現在存在するイベントを集約に変換するのに費やされる合計時間は15分です。

テスト中

毎朝、自動化されたテストを実行し、生データに関する質問、および集計の準備状況と品質に回答します。一週間前の同じ日に比べて。」



技術的には、これらはJUnitを使用し、ClickHouseのjdbcドライバーを実装する通常の単体テストです。 すべてのテストの実行はTeamCityで開始され、1スレッドで実行するのに約30秒かかります。失敗した場合、すばらしいTeamCityボットからVKontakte通知を受け取ります。

おわりに

ClickHouseの安定したバージョンのみを使用すると、髪が柔らかく絹のようになります。 ClickHouseの速度が低下しないことを付け加える価値があります 。