分断されたデータをひとつのデータプラットフォームに統合　LINEのIUに立ちはだかる3つの問題

本セッションがどのような人に役立つのか

奥田輔氏：LINE株式会社 Data Engineeringセンター、Data Platform室の奥田輔と申します。

このセッションでは「分断されてしまったデータを2,000台を超えるひとつのデータプラットフォームに統合した話」と題して、LINEにおけるデータプラットフォームがかつて抱えていたデータの分断という課題、それがどのようなものであったか、そしてその課題をどのように解決したかを紹介します。

内容に先立ちまして、本セッションがどのような人に役立つのかを紹介いたします。まずは、いわゆるデータエンジニアのポジションの方。弊社のデータプラットフォームはオンプレミスで巨大です。どのようなスケールなのか、またそれを支えるためにどのような構成になっているのか、LINEが技術的にどんなアプローチを採っているのかを知るのには、良い機会だと考えています。

次にデータを管理したい、利活用を促進したい方々。我々のデータプラットフォームは、LINEのすべてのデータを保存する、というビジョンを持っています。それゆえに、適切な権限管理や多くのステークホルダーをまとめ上げる力が必要です。

今回紹介するプロジェクトでも、さまざまなステークホルダーと関わりながら進行していきました。データの利活用を促進するには、どのようなアプローチがあるのか、その参考になれば幸いです。

最後に、データを活用してサービスや事業に日々役立てている方々。我々がデータを蓄積する目的は、そのデータを元に事実検証をし、サービスの改善をより素早く、より効率的に行うきっかけとして活用することです。Data Platform室という組織が、いったいどのようなことを考え、動いているかを深く知ることは、その活用の質を高めるという点において、非常に有意義だと考えています。

当然、上記に該当しない方々の参加も大歓迎です。それではさっそく本編を始めていこうと思います。

こちらが今日のアジェンダです。発表者とLINEにおけるデータプラットフォームをまず紹介して、本セッションで提起する課題、分断されたデータとはどのような問題なのか、そして、それを統合するのがなぜ難しいのかを説明いたします。

次に、その問題に対して我々LINEがどういったアプローチを採ったのか。技術的な観点と、データマネジメント観点の両方から説明いたします。

最後にそれらを解決した結果と、LINEのデータプラットフォームにおけるそこからの次の1歩を紹介して、本セッションの締めといたします。それではよろしくお願いいたします。

あらためまして、LINEData Platform室の奥田輔と申します。現在はData Engineering1チームにおいて、エンジニアリングマネージャーをしています。HadoopやElasticsearchといった分散システムの維持・運用や、Flink・Sparkなどを用いたインジェスチョンパイプラインの開発などを行う、データプラットフォームにおけるさまざまなエンジニアリングを担当するチームとなります。

2013年、新卒でLINEに入社して、LINE GAMEのDBAやデータ分析のETLの開発、パイプラインやHadoopの管理なども行ってきました。直近では、Hadoopの移行プロジェクトのリーダーも担当しました。こちらが、今回取り上げるプロジェクトとなります。

LINEのデータプラットフォーム「IU」

最初のセクションです。LINEにおけるデータプラットフォームを紹介いたします。このセクションでLINEのデータプラットフォームの構成、また、その規模感を簡単にお話しして、この後のデータ分断がどのようにして生まれたか、そしてそれがどのぐらいチャレンジングなものかをつかんでもらえればと思います。

我々はLINEのデータプラットフォームをInformation Universe、いわゆるIUと呼んでいます。LINEにおけるすべてのデータの蓄積、そして、それがさまざまなサービスへの効果的な利活用をもたらす、こういったビジョンを元に「情報の宇宙」と名付けました。これ以降のスライドでは、現在のLINEデータプラットフォームのことをこの名称、IUと呼びたいと思います。

さて、LINEのデータプラットフォーム、IUとは何か。みなさんが持っているスマートフォン上にあるLINEアプリやLINEのファミリーサービス、それらが通信するサーバーのログ、データベース情報など、LINEにおけるさまざまなサービスからのデータを蓄積しています。

データの蓄積だけでなく、これらを利活用するための解析、ダッシュボードツールや、システムにフィードバックするためのインテグレーション機能など、データの収集からその利活用までを一貫して行うプラットフォーム、それがIUです。

IUができた背景も簡単に紹介します。IUができる前は、組織ごとに分析環境を独自に持っていました。しかしながら、情報のサイロ化が起きてしまい、組織構造にとらわれない利活用が難しい状況だったのです。

そのため、LINEにおけるたった1つの分析環境として、データを使ったシステム連携のたった1つのエンドポイントとして、データ利活用のノウハウを提示するたった1つのスタンダードとして、それらのシングルを提供することで、よりシンプルに使いやすいデータプラットフォームにしたい。そして、LINEの行動規約の1つであるデータドリブンを加速させたい。その想いでIUは生まれました。

IUが持つデータパイプラインの全体像

こちらがIUが持つデータパイプラインの全体像になります。一番左側にデータ生成される各種サービスのシステムがあります。データの入口はKafkaを使用しています。そこからストリーミングフレームワークであるFlinkを利用し、Hadoop HDFSのみならずElasticsearchや外部システムへのリアルタイムフィードバックも行っています。

データベースは定期的なバッチ処理を、サービス側でUI上にて簡単に定義できるようにしています。これらの処理の多くはKubernetes上で動いており、サービスが増えていった時でも容易にスケールアウトできるようになっています。

Hadoop HDFSに蓄積されたデータは、さまざまなエンジンを使って解析されます。かつてはHiveがメインでしたが、最近ではSparkの比率がジョブ全体の半数を超えてきています。また、アドホックな統計処理はTrino、かつてはPrestoと呼ばれていたものを使って、より迅速にトライ・アンド・エラーできるようにしています。サービス連携用にはDatahubという、REST APIでデータをやり取りするためのツールを開発しています。

IUでは、データを触るユーザーがすべてエンジニアというわけではありません。「技術的なスキルを持たない、しかしながらサービスに役立つデータは見たい」というニーズも存在します。そのため、IUではさまざまなBIツールを使い分けています。

Elasticsearch用のKibanaや解析結果をダッシュボードとして見るTableau、Jupyterは有名かと思います。さらに、さまざまなニーズを満たすため、Prestoクエリを簡単に実行できるYanagishima、Sparkコードを抽出するとスケジューリングやビジュアライゼーションを行うノートブックシステムのOASIS。データを投入すると自動的に可視化まで行うLINE Analyticsなどを自分たちで開発しています。

これらデータを解析進める上で、適切な権限管理と運用の効率化は欠かせません。運用の効率化のために、設定管理ツールであるCentralDogmaを利用したり、GitHubと連携し、Prometheus、Grafanaを用いた監視・アラート体制を構築したりしています。

権限管理はRangerで統合的に行い、それをより高度にすべく、IU Webというカタログツールを開発して、データオーナーの定義などを行っています。

IUの規模感

ここでIUの規模を感じ取れる数字を見てみましょう。まずは、インフラの規模を示す数字たちです。HDFSのキャパシティは現在400PBを超えました。データプラットフォーム室で管理するサーバー台数は、Hadoop、Elasticsearch、Kubernetes、これらすべて合わせて5,000台以上です。YARNの規模感を表すvCore数は90,000に達しています。

次にデータの利活用の度合いを数字で見ていきます。IUではサービスが作り出すデータの単位をHiveテーブルで定義しています。そのデータの種類を表すHiveテーブル数は40,000を超えました。

1日当たりのYARN、Prestoなどのジョブの総数は150,000。実にさまざまな組織から、さまざまな用途でヘビーに使われています。そしてデータ流量のピークについて、これは秒間で17,000,000を超えることもあります。IUによって、LINEでは非常に大規模にデータ活用が進んでいることがわかってもらえると思います。

「分断されたデータ」とはどのようなものだったか

さて、LINEにおけるデータプラットフォーム、IUについて紹介しました。しかしながら、ここに到達する道のりは簡単ではありませんでした。それは、分断されたデータが存在したからです。このセクションでは、タイトルにあります「分断されたデータ」がどのようなものであったか、また、なぜ問題なのかを深掘りしていきます。

先ほどIUができた背景について触れました。IUができる前は、Datachain、Datalakeという2つの大きなクラスターが存在しました。それぞれ運用する組織が違っており、Datachainはメッセージ本体のデータ蓄積基盤として、Datalakeはファミリーサービスを含めた、より高度なデータ分析基盤として誕生しました。

しかしながら、それぞれのクラスターが成長するにつれて、その役割はより複雑に絡み合っていきました。「片方にないデータを結合してみたい」「出社時間帯にクラスターがビジーすぎるので別のクラスターを使いたい」……、次第に互いのクラスターを無理やり使い分けるユーザーが増えてきてしまいました。その結果、LINEのデータ分析におけるデータフロー、業務遂行がとても複雑になってしまったのです。

3つの問題「カタログ」「コンピューティング」「ストレージ」

問題を一つひとつ細かく見ていきましょう。大きく分けると3つの問題がありました。1つ目はテーブルカタログを持つMetaStore同士が分断されていることによる、テーブルのJOINができないこと。2つ目は、YARNやPrestoをそれぞれ運用することにより、互いにコンピューティングリソースの共有ができないこと。3つ目は、データを保存しているストレージの分断による権限管理の断裂です。

まずはカタログの問題、JOINができないことです。例を挙げて見てみましょう。このクエリは、LINEスタンプにおいて特に売れ行きが良いスタンプを送信しているユーザーとその送信数を取得するといった疑似クエリです。

ここでは2つのテーブルをJOINしていますね。しかしながら、この2つのテーブルは別々のクラスターにあったとしたらどうでしょう。HiveやSparkSQLでは現在1つのMetaStoreしかサポートしていません。つまり、別のクラスターにあるMetaStore情報とは一緒に見ることはできないのです。

これにより、分析効率がかなり落ちてしまったり、無理やりデータをクラスター間で交換したりする必要がありました。このデータの交換も後述する権限管理の観点などで、非常に運用コストが高いオペレーションでした。

次にコンピューティングの問題、リソースが共有できない、です。DatachainとDatalakeでは、それぞれのマシンの台数やそのスペックなどが、大きく異なっていました。

MLチームはDatalakeを使っていましたが、それゆえ常に計算リソースが足りない状況でした。一方でDatachainはデータ量こそ多いものの、計算リソースには余裕がある。リソース分配が不均等だったのです。

しかし「じゃあDatachainを減らしてDatalakeに組み替えよう」というわけにはいきません。HDFSはそれぞれに別々のデータを持っていますし、サポートしているツールも違います。

マシンを移行するにも、大規模なデコミッション作業にはある程度の時間が必要です。さらに、クラスターそのもののキャパシティプランニングも、ディスク、CPU、メモリ、それぞれの観点で行わなければいけません。これらを同時並列的に行うことは、かなりの労力が必要でした。

3つ目にストレージの問題、権限の断裂です。カタログの問題でお話ししたとおり、データをまとまって見るためにデータコピーが日常的に行われていました。しかしながら、データコピーを行うとHadoopそのものが変わってしまうわけです。そのため、それぞれのクラスターで権限体系が変わってしまう危険性がありました。

また、権限だけではありません。スキーマの変更やデータの削除ポリシーなど、データそのものの変更を行う時は、必ず両方のクラスターに同じポリシーを適用する必要があります。しかしながら、そのポリシー適用手順はクラスターごとに大きく異なったりするわけです。これは、データマネジメント的に大きなリスクであり、また、システム運用負荷を増大させる要因でもありました。

データ統合への課題

さて、ここまででデータの分断がいかなる問題を引き起こしたか、ご理解いただけますでしょうか。それでは、このデータの分断をどのように解決すればいいでしょうか。我々は、新しく1つのクラスター、現在でいうIUを作り、そこに統合することで解決しようとしました。しかしながら、これほど大規模なクラスターを簡単に統合できるものなのでしょうか。やはり、統合に関してもいくつの課題が見つかりました。

まず、どのようにデータをコピーするか、という課題です。Datachain、Datalake、それぞれにおいて、さまざまなユーザーが、さまざまな粒度でデータを作成しています。それらが一斉にデータコピーをしたらどうなるのか。当然、データの依存関係はグチャグチャになりますし、データを起因とするシステムは、さまざまなトラブルに見舞われるでしょう。

また、Hadoopクラスター間でデータコピーを行うdistcpでは、データの規模によっては多くのYARNリソースを消費します。このリソースはvCoreやメモリだけでなく、ネットワークトラフィック増も引き起こし、最悪の場合、データプラットフォームに関係ないサービスにまで影響を及ぼす危険性があります。

また、誰から最初に移行するのかも重要なお話です。LINEでは、分析ジョブ、テーブル間の依存関係が複雑に入り組んでおり、組織間でのデータのやり取りも頻繁に行われています。

そのため「依存元のデータを勝手に移行されてしまって、システムが動かなくなってしまった」というシナリオも十分考えられます。それぞれの組織ごとにコミュニケーションを行う方法も提案されましたが、かなりのコミュニケーションコストになる上、プロジェクト計画においても変更があった場合、それが容易ではありません。

なので、それぞれの組織が、ある程度自律的に移行作業を行っても、全体のデータフローには問題がないようにする必要がありました。

どこにアクティブなデータがあるのかも問題です。本移行プロジェクトの初期段階では、ストリーミングジョブを分割し、新旧それぞれにデータを書き込むDouble writeが提案されていました。しかし、この方法だと、書き込んだ両方のデータが必ず同一である、という整合性を満たす必要があります。

この要件は非常にハードルの高い方法であり、また、それをデータプラットフォーム室だけでなく、データを投入しているさまざまなサービス側にも担保してもらうのは、現実的に無理があります。

当然、管理するコンポーネントが増えれば、それだけコストも増えますし、何よりDouble writeの「どちらが正しいデータなのか」は事実上誰にもわからないケースが多いと考えました。そこで、Double writeではない、別の方法を考える必要が出てきたわけです。

さて、ここまででデータの分断、そして統合の難しさについてお話ししてきました。さまざまな問題・課題があることがご理解いただけたかと思います。そこで、我々はよりさまざまな状況に対してアプローチを考えました。次のセクションでは、それらのアプローチを紹介していきます。

（後半へつづく）