Hadoop とは大規模なデータを並列分散処理を行うフレームワークを提供 Google による MapReduce および Google File System(GFS) の論文をベースに開発された Apache プロジェクトの OSS MapReduce MapReduce 分散処理フレームワー

Size: px

Start display at page:

Download "Hadoop とは大規模なデータを並列分散処理を行うフレームワークを提供 Google による MapReduce および Google File System(GFS) の論文をベースに開発された Apache プロジェクトの OSS MapReduce MapReduce 分散処理フレームワー"

あいぞうすえたけ
5 years ago
Views:

1 超入門大規模分散処理フレームワーク Hadoop SRA OSS, Inc. 日本支社技術開発部エンジニア長田悠吾 Cloudera Certified Developer for Apache Hadoop Cloudera Certified Administrator for Apache Hadoop Copyright SRA OSS, Inc. Japan All rights reserved. 1

2 Hadoop とは大規模なデータを並列分散処理を行うフレームワークを提供 Google による MapReduce および Google File System(GFS) の論文をベースに開発された Apache プロジェクトの OSS MapReduce MapReduce 分散処理フレームワーク GFS HDFS 分散ファイルシステム Copyright SRA OSS, Inc. Japan All rights reserved. 2

3 Hadoop の歴史 2003 年 Google GFS の論文発表 2004 年 Google MapReduce の論文発表 2005 年 Hadoop プロトタイプの誕生 2006 年 20 ノードで動作 2006 年 Yahoo! が本格的に注力 2007 年 200 ノードで稼働 2008 年 Apache トップレベルプロジェクトへ昇格 2008 年大規模なソート処理で世界記録樹立 910 台のノードで 1TBデータをソート 297 秒 209 秒! 2008 年 11 月 Google 1TBデータをソート 68 秒大きな進歩企業によって 2009 年 1 月 Hadoop 1TBデータをソート 62 秒育てられたOSS 現在 Yahoo! 4000 台 Facebook 2250 台以上 Copyright SRA OSS, Inc. Japan All rights reserved. 3

4 Hadoop が注目された背景大規模データの保管と大規模データの分析時間の経過につれてデータ量が爆発バックアップできない! データの冗長性膨大な情報から価値ある情報を抽出するニーズ Webショッピングのリコメンダアクセスログ解析での行動分析大規模データの各データ間の分析マーケティングデータとして活用 Copyright SRA OSS, Inc. Japan All rights reserved. 4

8 Hadoop の得意不得意得意高スループット処理高スケーラビリティ高可用性安価なサーバで構成可能一般に調達できるサーバ不得意 RDBMSが得意なトランザクション一貫性保証データの更新 ( 出来ない ) OLTP 処理のような秒以下のレイテンシが必要な処理 (Hadoopの処理は分のオーダで処理が終わる ) Copyright SRA OSS, Inc. Japan All rights reserved. 8

10 Hadoop のファイルシステム HDFS: Hadoop Distributed File System 大規模分散ストレージ大規模データを複数のサーバに分散して保存クライアントからは 1 つのストレージとして見えるサーバを増やすことで拡張可能 (= スケールアウト ) HDFS Client... Copyright SRA OSS, Inc. Japan All rights reserved. 10

11 Hadoop のファイルシステム HDFS: Hadoop Distributed File System 通常のファイルシステム上で動作各サーバに保存されているデータは通常のファイル hadoop コマンド HDFS ext3 OS H/W 普通のファイルシステムと似たコマンドで扱える hadoop fs コマンド $ hadoop fs -put local_file.txt hdfs_file.txt $ hadoop fs -ls $ hadoop fs -mkdir mydata $ hadoop fs -cat hdfs_file.txt Copyright SRA OSS, Inc. Japan All rights reserved. 11

12 Hadoop のファイルシステム HDFS: Hadoop Distributed File System 冗長性ファイル A B C 1 つのファイルを複数のブロックに分割して同じブロックを複数のサーバーに複製 ( レプリケーション ) デフォルトのレプリケーション数は 3 サーバ 1 サーバ 2 アクセスパターンの制限 A B A B Write-Once C ファイルの更新は行わないシーケンシャルな読み込みを前提ブロックサイズ 64MB ランダムアクセスを想定していないサーバ 3 A C サーバ 4 B C Copyright SRA OSS, Inc. Japan All rights reserved. 12

13 HDFS クライアント HDFS メタデータのディスク書き出し Secondary NameNode 単一ストレージのようにファイルを読み書き可能ファイル 1 NameNode メタデータ A.. B b A B DataNode C : ファイル管理 & 監視 B C a ファイル 2 b.. a A.. C a DataNode b : DataNode 1 つのファイルを複数のブロックに分けてレプリケーション

14 HDFS の構成 DataNode ( スレーブノード ) DataNodes ファイルを構成するブロックを保存保持しているブロックがどのファイルを構成するか知らない NameNode ( マスターノード ) クライアントからのファイルの要求に応答メタデータを保持どのファイルがどのブロックで構成されるか NameNode 管理監視データ送受信どのブロックがどのノードにあるかファイル要求メタデータは基本的にメモリ上で管理し処理を高速化実際のデータの読み書きはクライアントと DateNode が直接通信 NameNode はボトルネックになりにくい Copyright SRA OSS, Inc. Japan All rights reserved. 14

15 HDFS の単一故障点 DataNodes マスタサーバの保護はしてくれない! NameNodeが故障するとHDFSが壊れる HDFSメタデータはオンメモリで管理メタデータ全体のスナップショット (fsimage) NameNode メタデータ管理監視 + 更新差分ログ (edits) をディスクに保存 SecondaryNameNode fsimage と edits をまとめて新しい fsimage メタデータのディスク書き出しを生成 ( チェックポイント生成 ) 複数ディスクに書き出し可能 NameNode が故障してもその替わりにはならない! Secondary NameNode Copyright SRA OSS, Inc. Japan All rights reserved. 15

16 HDFS の耐障害性 ~ データ損傷への対応 ~ レプリケーションファイルを構成するブロックはデフォルトで 3 つ複製される 1 つの DataNode が故障してもデータが失われることはないデータの読み書きの途中に DataNode で失敗しても他の DataNode に処理が引き継がれるクライアント側ではエラーの処理をする必要がない DataNode は NameNode に定期的にハートビートを送っている一定時間応答がない DataNode には障害が発生したとみなされる既定のレプリケーション数になるまで自動的に複製が行われるチェックサム各ブロックはチェックサム値を持ちデータ損傷の認識と修復が自動的に行われる Copyright SRA OSS, Inc. Japan All rights reserved. 16

17 Hadoop の並列分散処理 MapReduce 処理 ~ 概要例えば右のようなデータ ( ユーザ ID, 行動 ) から login したユーザの一覧が欲しい場合コマンドで実現するならば # usrid action login write_file logout read_file : $ cat input.txt grep login sort uniq > output.txt Map Map : 処理すべき値の生成 Shuffle & Sort Reduce Shuffle & Sort : 関連する値を一箇所に集めてソート Reduce : 集められた値の処理この Map 処理と Reduce 処理を複数サーバで並列に行う Copyright SRA OSS, Inc. Japan All rights reserved. 17

18 ジョブの依頼 Job クライアント JobTracker タスクの割り当てスケジューリング監視 TaskTracker TaskTracker TaskTracker Map Map Map Reduce Reduce Reduce Map / Reduce それぞれの処理を行うプロセスの生成と実行管理

19 Hadoop の並列分散処理 MapReduce 処理 ~ 詳細 Mapper 例 ) 転置インデックス行番号, その行のテキスト単語, 単語が現れる行のリスト Key-Value ペアからなるレコードを入力として受け取る新しく Key-Value ペアを生成して出力 ( 中間データ ) 例 ) 行番号, テキスト単語, 行番号 Shuffle & Sort 同じ中間 key に関連つけられた全ての中間データが集められる同じ中間 key を持つデータは全て同じ Reducer に渡される Reducer に渡される際には Key-Value リストはそのKeyの順序でソートされる Reducer Key-Value のリストを入力として受け取る Key の値に関してデータの集計を行い Key-Value ペアを出力例 ) 単語, 行番号のリスト単語行番号リスト( カンマ区切り ) Copyright SRA OSS, Inc. Japan All rights reserved. 19

20 Hadoop MapReduce のフレームワーク HDFS 上のファイルが入力 Mapper: 開発者が記述入力 :Key-Value ペア, 出力 :Key-Value ペア Shuffle& Sort HDFS 入力データデータ 1 データ 2 データ N Key に従ってデータを Reducer に渡すデフォルトでは Key のハッシュ値を用いるこのとき大量のネットワークトラフィックが発生! Key でソート Reduce: 開発者が記述入力 :Key-[Value list], 出力 :Key-Value Map Key = 1 のデータ Reduce Map Key = 2 のデータ Map Key = M のデータ Reduce 出力は HDFS 上のファイル Reducer の数だけファイルが生成される結果 1 HDFS 結果 M Copyright SRA OSS, Inc. Japan All rights reserved. 20

21 MapReduce の構成 JobTracker( マスタサーバ ) 依頼されたジョブをタスクに分割し TaskTracker への割り当て & スケジューリングを行うタスクの単位 1 つのタスクが 1 つの入力スプリット ( デフォルトではブロック単位 ) を処理可能な限り処理すべきデータを持っているノードへタスクを割り当てるデータがあるところにプログラムを持ってい = データローカリティ単一障害点 TaskTracker( スレーブサーバ ) 割り当てられたタスクの実行 Map プロセス Reduce プロセスの生成実行管理 JobTracker への進捗報告 Copyright SRA OSS, Inc. Japan All rights reserved. 21

22 ジョブの依頼 Job クライアント JobTracker タスクの割り当てスケジューリング監視 TaskTracker TaskTracker TaskTracker Map Map Map Reduce Reduce Reduce Map / Reduce それぞれの処理を行うプロセスの生成と実行管理

23 MapReduce の耐障害性 ~ 部分障害への対応 ~ TaskTracker は JobTracker に定期的にハートビートを送っている投機的実行ある TaskTracker が明らかに遅い場合同じタスクを別の TaskTracker に依頼処理の完了が早かった TaskTracker の結果を採用する一定期間応答のない TackTracker は強制終了され同じタスクが別の TaskTracker に割り振られる一定時間に多数の失敗を起した TaskTracker はブラックリストに登録される 1 つのタスクが失敗した場合でも全体のやり直しをせずにジョブを継続可能ただしタスクが 4 回 ( デフォルト ) 失敗した場合にはジョブ全体が失敗したとみなされ MapReduce ジョブは終了 Copyright SRA OSS, Inc. Japan All rights reserved. 23

24 MapReduce プログラムの記述 MapReduce APIの提供並列分散処理のフレームワーク障害処理等を意識する必要がない Map, Reduce の処理を記述するだけ記述は基本的に Java で行う Mapper クラス MapReduceBase クラスを継承 Mapper インタフェース (map メソッド ) を実装 Recucer クラス MapReduceBase クラスを継承 Reducer インタフェース (reduce メソッド ) を実装ドライバクラス Mapper クラス, Reducer クラス, 入出力ファイル形式の指定などジョブの設定ジョブサブミットその他 : Combiner, Partitioner, Comparator Copyright SRA OSS, Inc. Japan All rights reserved. 24

25 MapReduce プログラムの記述ジョブの実行 hadoop jar コマンドを使って jar ファイルと引数を指定 $ hadoop jar wc.jar WordCount inputfile outputfile Java 以外の方法 HadoopストリーミングAPI 標準入出力を介して map 処理 reduce 処理を行うより慣れた使い慣れた言語ライブラリ Ruby, Perl, Python, bash,... Hive, Pig SQL ライクな言語を用いてMapReduce 処理を記述できる Copyright SRA OSS, Inc. Japan All rights reserved. 25

29 Hive と Pig SQL ライクな言語でデータ操作を記述自動的にMapReduce 処理に変換され並列計算を行う Apache License Hive HiveQL: 宣言型 hive> INSERT OVERWRITE TABLE pv_users hive> SELECT pv.*, u.gender, u.age hive> FROM user u FULL OUTER JOIN page_view pv ON (pv.userid = u.id) hive> WHERE pv.date = ' '; Pig PigLatin: 手続き型 grunt> A = load '/etc/passwd' using PigStorage(':'); grunt> B = foreach A generate $0 as id; grunt> dump B; Copyright SRA OSS, Inc. Japan All rights reserved. 29

31 Oozie 複雑な処理を行うには複数の MapReduce をジョブを多段に行う必要がある例 ) ジョブ A とジョブ B の出力をジョブ C の入力とする Oozie : ワークフローエンジン Apache License 2.0 Hadoop のワークフローを実行ワークフローは XML で記述するジョブには Hive, Pig, Sqoop を含めることもできる Copyright SRA OSS, Inc. Japan All rights reserved. 31

32 Hadoop まとめ HDFS 分散ストレージ MapReduce 並列計算フレームワーク障害に強いシステムデータが失われない HDFS クライアント A B C : a b : HDFS 単一ストレージのようにファイルを読み書き可能ファイル 1 ファイル 2 メタデータのディスク書き出し NameNode メタデータファイル管理 & 監視 A.. DataNode C a Secondary NameNode A.. DataNode B b B DataNode 1 つのファイルを複数のブロックに分けてレプリケーション.. C b a ジョブの依頼 Job クライアント TaskTracker Map Reduce TaskTracker Map Reduce JobTracker タスクの割り当てスケジューリング監視 TaskTracker Map Reduce Map / Reduce それぞれの処理を行うプロセスの生成と実行管理 HDFS 入力データデータ 1 データ 2 データ N 処理が途中で止まらない Map Map Map システムが自律的に障害から復旧 Key = 1 のデータ Key = 2 のデータ Key = M のデータ Hadoop フレームワークがこれらの面倒を見てくれる開発者は MapReduce 処理の開発に専念できる Reduce 結果 1 HDFS Reduce 結果 M Copyright SRA OSS, Inc. Japan All rights reserved. 32

33 OSS プロフェッショナルサポートサービス 33 種類以上の幅広い OSS をワンストップでサポート台数無制限のサービスですサービス内容ヘルプデスク障害対応ナレッジサービス情報配信サービス ( プロメニューのみ ) サービス対象ソフトウェア FTP サーバ ProFTPD,vsftpd ロードバランサ / リバースプロキシ Pound ファイル / プリントサーバ Samba キャッシュサーバ Squid DNS サーバ Bind LDAP サーバ OpenLDAP メールサーバ Postfix,sendmail qmail,dovecot UW-IMAP Courier-IMAP Qpopper 運用監視 :Hinemos Zabbix KVS: memcached Kyoto Cabinet,Kyoto Tycoon 分散処理 :Hadoop シングルサインオン OpenAM Web サーバ :Apache AP サーバ :Tomcat DB サーバ :PostgreSQL SQLite HA ソフトウェア : Heartbeat, Pacemaker, DRBD 仮想化 : Xen,KVM OS: CentOS

35 MapReduce 例 : ワードカウント <Key = *, Value = テキスト > <Key= 単語, Value = 単語出現回数 > Map (Key, Value) { foreach (word in Value) { output (word, 1); } } テキストを単語に分割して単語と整数 1 を出力 Reduce (Key, Values) { count = 0; 各単語について出現回数を合計 foreach (value in Values) { count += value; } output (Key, count); } Copyright SRA OSS, Inc. Japan All rights reserved. 35

36 MapReduce 例 : 転置インデックス作成 <Key = 行番号, Value = テキスト > <Key = 単語, Value = 単語を含む行番号リスト > Map (Key, Value) { foreach (word in Value) { output (word, Key); } } テキストを単語に分割して単語と行番号を出力 Reduce (Key, Values) { list = list (Values); output (Key, list); } 各単語について行番号のリストを出力 Copyright SRA OSS, Inc. Japan All rights reserved. 36

37 WordCount: Mapper public static class Map extends MapReduceBase 入出力 Key-Value の型を指定 implements Mapper<LongWritable, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); } } public void map(longwritable key, Text value, } OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { String line = value.tostring(); StringTokenizer tokenizer = new StringTokenizer(line); while (tokenizer.hasmoretokens()) { word.set(tokenizer.nexttoken()); output.collect(word, one); map メソッドの実装 Hadoop の型 IntWritable LongWritable Text : Copyright SRA OSS, Inc. Japan All rights reserved. 37

38 WordCount: Reducer public static class Reduce extends MapReduceBase 入出力 Key-Value の型を指定 implements Reducer<Text, IntWritable, Text, IntWritable> { public void reduce(text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { int sum = 0; while (values.hasnext()) { reduce メソッドの実装 sum += values.next().get(); } output.collect(key, new IntWritable(sum)); } } Copyright SRA OSS, Inc. Japan All rights reserved. 38

39 WordCount: ドライバ public class WordCount { public static void main(string[] args) throws Exception { JobConf conf = new JobConf(WordCount.class); conf.setjobname("wordcount"); conf.setoutputkeyclass(text.class); conf.setoutputvalueclass(intwritable.class); ジョブの設定出力 Key-Value の指定 conf.setmapperclass(map.class); conf.setcombinerclass(reduce.class); conf.setreducerclass(reduce.class); conf.setinputformat(textinputformat.class); conf.setoutputformat(textoutputformat.class); Mapper, Combiner, Reducer の指定入出力形式の指定 FileInputFormat.setInputPaths(conf, new Path(args[0])); FileOutputFormat.setOutputPath(conf, new Path(args[1])); 入出力ファイルパスの指定 } } JobClient.runJob(conf); ジョブのサブミット Copyright SRA OSS, Inc. Japan All rights reserved. 39

スライド 1

スライド 1 Hadoop とは Hadoop の二本柱分散ファイルシステム HDFS(Hadoop Distributed File System) 分散プログラミングモデル MapReduce Hadoop の目的大規模ファイル処理格納, 加工ペタバイト規模複数計算機の協調動作スケーラブルシステム数百 ~ 数千台規模 Key-Value store MapReduce HDFS Node Node

Hadoop とは 大規模なデータを並列分散処理を行うフレームワークを提供 Google による MapReduce および Google File System(GFS) の論文をベースに開発された Apache プロジェクトの OSS MapReduce MapReduce 分散処理フレームワー

Hadoop とは大規模なデータを並列分散処理を行うフレームワークを提供 Google による MapReduce および Google File System(GFS) の論文をベースに開発された Apache プロジェクトの OSS MapReduce MapReduce 分散処理フレームワー