Hadoopエコシステムをざっくり整理してみる #hadoop

Hadoopエコシステムをざっくりと総整理(2018)

CL; TR;

Hadoopという大きなブレークスルーに群がるようにできたエコシステムたちを整理してみる
本記事では難しいお話は抜きにして概要のみさらっと流す

What is Hadoop?

分散システムをめちゃくちゃ簡単に構築できるようにしたフレームワーク（簡単になったとは言ってない）

ちなみに分散システムとは、

複数のサーバを連携させて一つのサーバとみなす
なんからの計算処理を複数のサーバで手分けして行う
- Ex. 夏休みの宿題を家族で手分けすれば早く終わりますよね、ね！
高価かつ超ハイスペックサーバじゃなくてもコモディティハードウェアを複数台束ねるだけで、構築可能(スケールアウト)
リソースが足らなくなってもサーバを足せばよい(既存サーバを有効活用できる)

といいことがいっぱいあるのですが、問題もあって、
本来分散システムを組む場合、

サーバが壊れたときにそこで処理していたものはどうする?
サーバが問題なく動いているかを誰が監視するの?
データが来ないんだけどこれってネットワークの問題?それともサーバの問題?

などなど、、、考えることが山積みなのです。。。

※余談
分散システム関連は研究の世界でもいろんな問題があります。
合意問題, リーダ選挙問題, 分散スナップショット, ノードの故障検知とかとか・・・
おっと此処から先は闇だ
（自分もこのあたりはこれから勉強していきたい）

で、それまで分散システムを組む場合は上記のような問題を考慮したアルゴリズムを実装する
必要があるのですが、なかなか難しいわけです。

そこで、難しいお話をフレームワークに落とし込むことで、
分散システムのアルゴリズムを１から実装するとこなく、分散システム上でデータを保存したり処理できるようにしたのがHadoopなのです。

ちなみにHadoopは大きく分けて3つのコンポーネントがあります。

コンポーネント	概要
HDFS	大きなサイズのデータを分散システム上で保存するファイルシステム
MapReduce	分散システム上でデータを処理するためのプログラミングモデル
YARN	MapReduceを始め、後述する分散処理フレームワークによるデータ処理のマネージングを行う（どのサーバが暇しているとか）

で、ここからが本題。

Hadoopができました! → これで分散システムが組める!!Yappy!!!!! →　こんな機能がほしいな〜あんな機能がほしいな〜　→　自分たちで作っちゃえ!!
ってな流れでHadoopのWeak Pointをカバーするアプリケーションが多く誕生しました。（多分）
それらのコンポーネント郡を総称してHadoopエコシステムと呼びます。

ちなみにHadoopのweak pointは大きく分けて以下にカテゴライズされます。

プログラミングやりたくない問題
バッチ処理問題
- データの抽出問題 → 一部のデータだけ変えるとかができない。小回りがきかない。
- スピード問題　→とにかく遅い。
- IoT等によるストリーミング問題　→　リアルタイムにデータをさばけない。
その他

プログラミングやりたくない問題

Hadoop上で並列分散処理をする場合、基本的にはJavaのフレームワークが提供されているので、
JavaでMap処理、Reducer処理・・・とかを書いていくのですが、それがめんどう。
というよりビッグデータを扱う人はどちらかというとエンジニア < アナリストであるので、
そういった層も分散処理できるようなインターフェースの開発が盛んになりました。
その代表例がSQL。

MapReduce処理をSQLで記述できるHive

HiveはFacebookによって開発されたSQLインターフェースのようなものです。
Hiveは一般的なRDBのようにテーブルを作成でき、
SELECT 〇〇 FROM △△ というSQL文を記述することでデータを抽出できます。

ほんとのRDBとの違いは裏側の処理エンジン。
Hiveでは処理エンジンはMapReduceのままです。詳しく説明するとSQLを実行する差異、HiveがSQL構文を解釈しMapReduce処理としてHadoop上で実行されます。

そのため、RDBのように短時間でデータは返ってくることはありません。
SQLだけどバッチ処理

なので、Hiveそのものを使うことは最近ほとんどありませんが、後述するSQL用の処理エンジンでもテーブル構造を格納するメタストア機能やサーバ機能はHiveであるケースが多いです。

スクリプト処理でMapReduceを記述するPig

Pigはヤフーが開発した、Hadoop上で処理可能なスクリプティング処理のインターフェースです。
Pig Latinという独自のスクリプト言語を実行することができます。
Javaに比べると学習コストも低いため実装しやすいと言われてました。
ただ、Hive(SQL)のほうが有名なのであまり使われているケースを聞くことは自分はないです。

バッチ処理問題

MapReduceはそもそもバッチ処理を前提としたプログラミングモデルです。
ちなみに処理方法を大別すると、

処理の一覧	詳細
バッチ処理	処理実行前の準備に時間がかかる(レイテンシが大きい)ものの、大量のデータを一気にさばける。数分〜数時間単位
インタラクティブ処理	処理を実行すると数秒から数分の間に結果が返ってくる。ただし大量のデータをさばくのは難しい。一般的なRDBのSQLはこれに属する
リアルタイム処理	随時くるデータを数ミリ秒単位で処理する。

Hadoopは大量にデータをさばくことと分散ノード上にあるデータに対して一部のノード停止によるデータ紛失をさけるため、耐障害性を加味したアーキテクチャになっています。そのためバッチ処理によったアーキテクチャになっているのですが、それによって以下の部分が犠牲となっています。

弱点	内容
小回りがきかない	データの更新やフィルタリングして抽出みたいなレコード単位の処理が難しい
スピードが遅い	耐障害性を保ちつつ、より高速に処理ができるエンジンがほしい
リアルタイムにデータがさばけない	大量のデータがIoT等で取得できるようになったものの、リアルタイムに来るデータをさばいて可視化や通知機能といったことができない

ランダムアクセス可能なHBase(小回りきかない問題)

HBaseもHadoopエコシステムの一つです。
HBaseの特徴はRDBライクなテーブル処理ができ、データの一部だけを取得・挿入・更新等ができるランダムアクセス性を持っています。
またRDBのように事前にスキーマの定義(DDL)を設定する必要がないため、データレイク的な役割を果たすことができます。
データ構造としてはワイドカラム型という構造になっています。

ワイドカラム型とはなんぞや?という話からすると、キーバリュー型 + 横の列をほぼ無限にしたという形です。
なので、

key	val1	val2	val3
1	aa		bb
2		cc	dd

みたいな形になります（実際のデータ保存方法は少し違いますけど）
また、カラムのグループとしてカラムファミリーが作れます。

なので、データの格納には、

テーブル名
カラムファミリー名
カラム名
キーの値

でデータを入れます。
※同じテーブル内の異なるカラムファミリー名に対して、同じカラム名を指定することは可能

※このあたりのデータ構造の詳細はNoSQLというジャンルで記事や本を読まれるとよいかと思います。

またこれらのデータは、裏側ではHDFSに保管されているため耐障害性もあります。
また、テーブルはノード単位ではなくRegionというブロックのような単位で扱われます。
なので、

Regionが膨らんできたらRegionそのものを分割する
ノードごとにRegionの大きさが偏ってきたらロードバランシング（Regionの大きさを均一にするように調整する）

といった機能ももっています。
使用用途としてはとくにメッセージ系のDBとして使われるケースが多いですね。

HDFSとHBaseのいいとこどり?? Kudu

KuduはHDFSとHBaseのいい部分をとった分散ストレージサービスと言われています。
具体的には、

主にデータ抽出(SELECT処理)に特化したDWH型のストレージ
RDBのようにスキーマを決めておく必要あり
SQLインターフェースがあるためクエリでデータを取得可能

※もうちょっと勉強して足す予定。。。

Hiveと連携できる処理エンジン

HiveがでたことでSQLで分散処理が書ける!!!となったものの問題発生。

SQLで書けるけど・・・結果返ってくるん遅い!!!

一般的なRDBだったらクエリ実行から遅くとも数分くらいで返ってくるはずなのに、Hiveだと数時間たっても返ってこない・・・なんてこともあります。
それもそのはず、Hiveの処理エンジンはMapReduce = バッチ処理なのでRDBみたいなインタラクティブな処理はできません。

でもそれで諦めません。

「早く走るエンジンを作ればいいんじゃね?」

そこでMapReduceに変わる並列分散処理エンジンが台頭してきました。

コンポーネント	概要
Tez	MapReduceより早い。Hiveインターフェースと組み合わせて数倍〜数百倍のパフォーマンスで分散処理ができる
Impala	Cloudera(Hadoopディストリビューション屋さん)が開発。Hiveのメタストア等と連携する。
Spark SQL	分散処理フレームワークであるSparkの一機能。Impalaと同じくHiveと連携して処理を行う。
Presto	Facebookが開発。もちろん早い。HDFS以外のデータソース(RDB等)との接続も可能

※このあたりの細かい違いがよくわかっていないです。。。

最近よく聞くのはPrestoかSparkSQLですかね。

MapReduce2.0ともいうべき汎用的な分散処理エンジンSpark

最近お金2.0とかモチベーション3.0とか、〇〇2.0的なの流行ってますよね。
そういう意味ではMapReduce2.0といっても差し支えないのがApacheSparkです。
SparkはUCバークレー(アメリカでは有名なコンピュータサイエンスに強い大学です)のAMPlabにて開発された分散処理フレームワークです。

MapReduceとの大きな違いは

インメモリ処理
- 基本的にデータをインメモリに入れたまま処理するためMapReduceよりディスクI/Oが少ないので早い。
遅延実行(DAG(非循環有向グラフ))
キャッシュ機能による反復処理の効率化
- 効率的な処理が可能になる
バッチ処理だけでなく、リアルタイム処理、機械学習、SQLライクな処理のライブラリがある

というあたりでしょうか。

今なお、DataFrameやDataSet等効果的な処理という部分で開発が進められており、
分散データ処理といえばほぼデファクトスタンダードになっています。

リアルタイムでデータを送るFlume,Kafka

最後はリアルタイムでデータを送るコンポーネントの紹介です。
特にIoTの世界ではリアルタイムで機械の状態をチェックしたり、スパイク(急に値が大きく変わる現象)をウォッチしたいニーズもあり、最近リアルタイムのデータ処理が増えてきているのもあり、様々なOSSがでてきています。

結構リアルタイムのデータ処理アーキテクチャは一見難しそうに見えるのですが、登場人物を整理してみると、

基本的には

やってきたデータを一時的に保管する場所(簡易的にストアと呼ぶようにします)
ストアにデータを送る
ストアからデータを取得する

しかありません。

この組み合わせにプラスして、
元データ→データ一時保管→データの保管先
のそれぞれの矢印に対して、

随時データ元がデータを送るのか（Push型）
一定間隔でデータを取ってくるのか（Pull型）という違いがあります。

KafkaならびにFlumeの場合、

元データ(ログ等)からデータの一時保管に随時データを送信する（Push型）
最終のデータ保管先(HDFS等)が定期的に一時保管先にデータを取りに行く(Pull型)

という構成になっています。

また、Kafkaの場合一時保管する場所(Broker)をクラスタリングし、かつレプリケーションすることができるのでメッセージの耐障害性を確保できます。
※さらにクラスタを複製するMirror Makerという機能があったり。。。

その他

RDBからHDFSにデータを送るSqoop

HDFSはファイルシステムです。
しかしながら、基幹システムのデータ等はRDBだったりするケースが大半だと思います。
その場合、HDFSにデータを送るためには、

RDBからデータを取り出してCSV等テキストファイル化する
テキストファイルをHDFSに配置する

というステップが必要になるのですが、それをコマンドだけでうまいことやってくれるのがSqoopです。
またSqoopはバルクロードに対応しているので大量データも高速に転送できます。
さらに、RDB→HDFSだけじゃなくて、HDFS→RDBにデータを転送できるのも強みです。

ファイルを効率的に持つAvro,Perquet

HDFSはファイルシステムです。(2回目)
HDFS上にはテキストファイルを配置することになるのですが、ファイルはレプリケーションされることもあるので、
できる限りデータを圧縮してもちたい、けど処理実行時の解凍によるレイテンシは嫌だという声があるわけです。
そこでデータ処理に適したファイルフォーマットとしてAvro,Perquetが登場してきました。

両者の共通点としては

データをシリアライズできる
バイナリ形式
Snappy等の圧縮アルゴリズムを使って、ブロック分割したときにスプリッタブル(ファイル分割状態のままでも処理可能)なので分散処理に適している

があります。

両者の違いは、

Avro

行指向
スキーマ（カラム）構造の変更に強い

Perquet

列指向
- 一部のカラムのみ読み出すときのスピードが早い
- Avroよりデータのファイルサイズが圧縮される

が挙げられます。

分散処理のワークフローを定義するOozie

分散処理は単一のアプリケーションで終わることはあまりなく、
基本的には複数の処理を順番に実行して要件を満たす処理を実装します。
そこで、複数の処理をワークフローとして管理できるのがOozieです。

Oozieはワークフロー(Job)を構築できるだけでなく、

Jobの監視
HiveやMapReduce,Sparkとうことなる処理を単一のJobにまとめられる

といった利点もあります。

分散のノードの協調管理をしてくれるZookeeper

分散システムを構築する際、
クラスタリングされている各ノード間を協調させることが必要です。
具体的にどんな協調をするかというと、、、

自分以外にどのノードが分散システムに組み込まれているのか
今動いていないノードがあるか（どのノードが動いていないか）
複数のノードが動いているとまずいときに、他のノードの動きをロックさせる
設定ファイルを全ノードで共通化させる

などなど。

こういったことをノード間（それこそ100ノード以上とか）でやるのは非常に大変です。
そこで、分散システムの協調管理をしてくれるマネージャのような存在がZookeeperです。

実際にHBaseやKafkaではZookeeperの使用を強制あるいは推奨されています。

終わり

Hadoopのエコシステムをざっくりと整理してみました。
こうみると技術の進歩は全くの無からではなく、既存の課題解決から生まれていくんだなとしみじみしました。
（まさに巨人の肩の上に乗る）
もっともっと分散システム界隈の勉強をしていこうと思います。