現在JavaScriptの設定が無効になっています。このページを利用するにはJavaScriptの設定を有効にしてください。2019の16日目の記事です。一覧はこちら(外部リンク)こんにちは。ヤフーの橘(@moja_0316)です。私は2018年に新卒でデータ統括本部に入社し、データパイプライン領域でエンジニアとして働いています。今日は皆さんにヤフーのデータパイプラインの役割と、私たちが取り組んだデータパイプラインの信頼性を高める取り組みについてご紹介します。ヤフーのデータパイプラインヤフーは検索やEコマース、ニュースをはじめとした多くのサービスを運営しています。それらのサービスが保持するデータは非常に量が多く、かつ価値の高いものです。近年はデータソリューションサービスをはじめとして、さまざまなサービスのデータを横断して適切に利活用することで皆様の生活をより便利にする取り組みを多く始めています。サービスのデータを横断的に利活用するためには、まず各サービスのデータを効率的かつ簡単にデータレイクと呼ばれる場所に収集したり、さまざまなシステムとデータを受け渡したりする工程が発生します。以下の図のようにこの工程を担い、効率的にデータをやりとりするシステムをデータパイプラインと呼んでいます。データパイプラインを作る場合、メッセージングキューイングシステムと呼ばれるソフトウエアを活用して作る例が多いです。KafkaというOSSをベースにデータパイプラインを構築しています。Kafkaとその一部の概念をご紹介します。Kafkaは、Publish/Subscribeアーキテクチャを採用した分散メッセージングシステムと呼ばれるOSSです。KafkaではConsumeと呼ばれる)するConsumerと、中央でメッセージキューの役割を担う複数台のBrokerで構成されます。いう概念が存在します。Topicとはメッセージの種類を示す概念です。Bに...」というように異なるカテゴリーのメッセージを1つのKafkaクラスタで扱うことができます。PartitionとはKafkaに実際にTopicのメッセージをキューイングする時の並列処理数を表す概念です。ここまでの内容を図に示すと以下の通りです。PartitionはKafkaでの処理の分散やメッセージの保持を行うために非常に重要な概念です。一般的に1つのTopicは複数のPartitionに分かれており、Producerは一定のルールに従って複数のPartitionにメッセージを振り分けながらProduceします。Partitionはメッセージの欠損を防止するため以下の図のように複数のReplicaというキューで構成されており、名前の通りReplica同士でメッセージを複製しながらキューイングしていきます。以下の図の例では2つのPartitionがそれぞれ3つのReplicaで構成されています。同期しているReplicaがあれば、片方のReplicaがダウンしても全てのメッセージを保持したReplicaが他に存在しますが、同期したReplicaがない状態で全てのメッセージを保持したReplicaがダウンするとメッセージを失ってしまいます。Replicaの存在するマシンに故障が発生してもメッセージが欠損しない状態を担保して稼働できます。それ以上の故障が発生した場合、Partitionは「欠損する可能性を受け入れながら稼働を続ける」か「Replicaが復帰するまでメッセージがProduceされることを拒否する」のいずれかの動作を行います。0はメッセージの欠損もしくはProduceの停止が発生する可能性があります。メッセージの欠損防止と可用性を低コストで両立するためにメッセージ欠損の発生シーンと一般的な対策サービスのデータをより横断的に利活用できる環境は、いつでも確実に必要なデータが用意できなければ成り立ちません。データを集める役割を担うデータパイプラインの品質はこのような環境の品質に大きく影響を与えます。そのため私たちはパイプラインの品質の目標として「メッセージを始点から終点まで欠損なく運びつづけること」「パイプラインが常に稼働を続けていること」の2点を掲げています。2)台を超えるサーバーの故障を同時に起こしてはいけません。複数のサーバーが同時多発的に故障することはめったに発生しませんが、万が一発生してしまった時の影響の大きさを考えると、これを阻止する方法を構築する必要がありました。単純なのは、「Replica数を増加させる」という方法になると思います。Replica数を増やすことはサーバー台数を増やすこととほぼ同義であるため、相応の設備投資が必要になり経済合理性を得るのが難しくなります。Replica数を増やすということは、その分だけReplica間の複製処理やトラフィックの負荷が大きくなるということであり、クラスタ全体へのパフォーマンスにも影響を与えかねません。2)台を超えるサーバーの故障が発生した場合Replica数が1以下になってしまい、メッセージの欠損もしくはProduceの停止が発生しうる」という状況の打破には至りません。加えて、私たちの要件を満たし解決の参考になる他の事例は見当たりませんでした。いう構成をソフトウエアで作り上げることができないか考えました。~先ほど「Producerは一定のルールに従って複数のPartitionにメッセージを振り分ける」と書きました。この振り分ける動作で重要なProducerのコンポーネントをPartitionerと呼びます。アプローチでこの課題を解決することにしました。Partition」に振り替えてProduceします。この振り替えてProduceする動作を私たちはFailoverと呼んでいます。上の図はFailoverの概念を簡潔に表した図です。Replicaが3つ存在する環境で2台のマシンが故障し、Partition0のReplicaは1つしか生存していない状態に陥ったとします。Partitionerはこの状態を検知し、Replicaが2つ以上生存しているPartition1にProduce先を振り替えてメッセージを送信します。PartitionにメッセージをProduceすることがなくなるため、本質的にメッセージの欠損を抑えながら稼働を続けることが可能になります。理由などの詳細は省きますが、この考え方でFailoverを行うためには以下の条件をクリアする必要があります。Partitioner)の導入を決定し、実装を進めることができました。実装において苦労した点この実装において最も苦労した部分は、不良を検知して切り替える瞬間のメッセージを正しくFailoverさせる制御の難易度が非常に高かった点です。Kafkaはいわゆるストリーミング処理を担うシステムですので、メッセージは常に流れ続け、途切れる瞬間はありません。Partitionが実際にOffline状態になってから、ProducerがそのPartitionをOfflineになったと認識するまでにはタイムラグが発生してしまいます(詳しくは省略しますが、メッセージをBrokerにProduceするタイミングとPartitionの情報をBrokerに取得するタイミングは異なるため、数秒程度のタイムラグがあります)。OfflineになってしまったPartitionにProduceしようとしてしまうメッセージがかなりの数発生してしまいます。加えてKafkaは正常な状態でないPartitionにProduceを試みた場合エラーを出して一定回数Produceを再試行しますが、再試行の際にメッセージを別のPartitionに割り当て直すことは行えない構成となっています。そのためPartitioner以外に、エラーを起こしたメッセージを失敗用キューに一度ため込み、再び新たなメッセージとしてProduceを行うためのリトライ用コンポーネントを新たに設計/実装し、図のようなアーキテクチャにすることになりました。ここからは少々実装寄りの話になりますが、私たちはKafkaのProducerをsaramaというGolang製のクライアントライブラリを利用して作成しています。saramaではKafkaをProduceするメソッドはgoroutineとchannelを用いて実装されています。そのため失敗したメッセージを受け取りそのままProduceするchannelを作ってしまうと、メソッドのoutputがそのまま自身のメソッドのinputに直接つながり、goroutineがデッドロックの状態に陥ります。これを防ぐためchannelにBufferを持たせることになりますが、Producerに流れるメッセージの流量やReplica数の変化を認識するまでのタイムラグは毎回異なるため、必要十分なBufferのサイズを見積もることが困難でした。考えられるBufferサイズの最大値を設定した場合、Failoverする数秒間のために、ある程度まとまったメモリ容量を確保することになり、あまり許容できるものではありませんでした。そこで私たちは失敗したメッセージが大量にあると判断した場合、予め新たなメッセージのProduceを停止する制御を追加しました。Counterで保持しておき、2つの差分を比較することにしました。2つの値に大きな差分があった場合は失敗したメッセージが大量に存在すると判断し、以下の図のようにMergerと呼ぶコンポーネントで失敗したメッセージのみをProduceするように制御しました。この制御を追加し、Buffer溢れを防ぐことでFailoverを実行する瞬間に発生するProducerのデッドロックを防ぎました。Partitioner自身の改良と合わせてこの改良を取り込むことで、欠損とパイプラインの停止を最小限に抑制するProducerを完成させることができました。Partitioner実装の取り組みについてご紹介しました。この機能追加を行ったProducerは、テストを経て今年の3月ごろに社内リリースを実施し、サービスのサーバーに多数導入され、私たちのデータパイプラインの品質向上に貢献しています。今回の例のようにヤフーのデータプラットフォームでは、万が一の事態を考えて先回りして品質向上に取り組んでおり、今後も積極的に行っていく所存です。データパイプライン領域は他のビッグデータ処理領域と比べて認知度などは決して高くないですが、この記事をきっかけにデータパイプラインやその周辺の技術に興味を持ってくれる方が増えればうれしいです。Foundationの米国およびその他の国における登録商標または商標です。Kafkaを用いたデータパイプラインの運用や開発を主に行っています。Uの運営もやっています。JAPANでは情報技術を駆使して人々や社会の課題を一緒に解決していける方を募集しています。詳しくは採用情報をご覧ください。JAPANのサービス開発を支える技術やデザイン、文化を発信しています。Reserved.このページの先頭へ。