Spark と大規模データ処理 - NAISTビッグデータアナリティクス第2回

Size: px

Start display at page:

Download "Spark と大規模データ処理 - NAISTビッグデータアナリティクス第2回"

かずまさかんけ
5 years ago
Views:

1 Spark NAIST

2 Hadoop Spark Apache Spark 2 / 39

3 (three V) (Volume) ZB ZB 1ZB 1 = B = 1, 000 EB = 1, 000, 000 PB = 1, 000, 000, 000 TB = 1, 000, 000, 000, 000 GB (Variety) Word, Excel (Velocity) / 39

4 4 / 39

5 Relational Database Management System(RDBMS) Oracle, MySQL, PostgreSQL, MicrosoftSQL,... SQL = or = 5 / 39

6 ACID Atomicity : Consistency : Isolation : Durability : (Consistency) 6 / 39

7 CAP (Consistency) (Availability) (Partition-tolerance) RDB KVS (Key Value Store) 7 / 39

8 KVS Key Value SQL = NoSQL Not Only SQL NO SQL SQL Value 8 / 39

9 KVS Google BigTable ( ) Google (Google Reader, Google Map, Blogger,...) Google BigTable Apache HBase - Hadoop KVS Apache Cassandra MongoDB Neo4j - KVS... 9 / 39

10 Standalone SQL : Oracle MySQL : PostgreSQL : SQLite NoSQL : MongoDB Cassandra : 10 / 39

11 Java Map Perl Hash 11 / 39 Key Value Store JSON JSON {"Key" : "Value"} {"Key" : {"Key2": "Value2"} } [{"Key":"Value"}, {"Key2": "Value2"},...] {" ": {" ":"10kg", " ":"A4", " ":"10 "} }

12 / = = SPOF (Single Point Of Failure) 12 / 39

13 (Java like language) File file = fopen("file.txt"); // int count = 0; // count for(string s: file){ if(s.contains(" ")){ // count += 1; // count 1 } } System.out.println(" "+count+" "); count 13 / 39

14 CPU 1 CPU 200MiB/s 380 IOPS (SAS 15000rpm) 14 / 39

15 A B / 39

16 MapReduce (Google) (map) map (reduce) MapReduce Google Hadoop Google File System GFS Google hdfs Hadoop Distributed File System 16 / 39

17 サーバの変化 *1 一つの大きなサーバ小さなサーバの集合体ブレードサーバ *1 by Robert Kloosterhuis (Wikimedia Commons) 17 / 39

18 京一つひとつは小さなコンピュータ上の方 12 個のサーバ 2 中央は電源下の方 12 個のサーバ複数のサーバを効率良く組み合わせられるかが勝負京のラック 2 by CES1596 (Wikimedia Commons) 18 / 39

19 Hadoop Spark ( 1) Hadoop Spark ( 2) 19 / 39

20 Hadoop ( ) import java.io.ioexception; import java.util.stringtokenizer; import org.apache.hadoop.* public class WordCount { public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasmoretokens()) { word.set(itr.nexttoken()); context.write(word, one); } } } public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static void main(string[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setjarbyclass(wordcount.class); job.setmapperclass(tokenizermapper.class); job.setcombinerclass(intsumreducer.class); job.setreducerclass(intsumreducer.class); job.setoutputkeyclass(text.class); job.setoutputvalueclass(intwritable.class); 20 / 39

21 Hadoop Map Reduce Map Reduce Map Reduce 21 / 39

22 Apache Spark Scala, Java, Python, R shell SQL Java JVM NAIST ITC 22 / 39

23 Apache Spark 1 bda1node???? Spark hdfs 3 python Java/Scala 4 23 / 39

24 PySpark %, $ pyspark $ ssh bda1node05.naist.jp % pyspark Spark Python (r266:84292, Jan , 01:49:05) Using Python version SparkContext available as sc. >>> exit() Spark Ctl-D % logout 24 / 39

25 % ls grep txt iphone6.ja.txt.xz iphone6s.ja.txt.xz Apple.txt.xz.ja.txt.xz.ja.txt.xz // xz // % xz -d Apple.txt.xz // % ls grep txt Apple.txt 25 / 39

26 hdfs % ls grep ja.txt iphone6.ja.txt iphone6s.ja.txt.ja.txt.ja.txt //.ja.txt spark % hadoop fs -put.ja.txt // % hadoop fs -ls Found 2 items drwx ysuzuki is-staff :14.Trash -rw-r--r-- 3 ysuzuki is-staff :04.ja.txt 26 / 39

27 % pyspark // >>> tweets = sc.textfile(u.ja.txt ) // >>> nara = tweets.filter(lambda s: u in s) // >>> nara.count() 15/10/06 15:25:03 INFO DAGScheduler: Job 3 finished: count at < 2640 // / 39

28 RDD (Resilient Distributed Dataset) ( ) ( ) DB RDD RDD ( RDD) PairRDD ( Key Value RDD) SchemaRDD (SparkSQL RDD) 28 / 39

29 % pyspark // >>> tweets = sc.textfile(u.ja.txt ) // >>> nara = tweets.filter(lambda s: u in s) // >>> nara.count() 15/10/06 15:25:03 INFO DAGScheduler: Job 3 finished: count at <stdin>:1, took s 2640 // / 39

30 // (--master local[2]) % pyspark --master local\[2\] // 15/10/06 15:25:03 INFO DAGScheduler: Job 3 finished: count at <stdin>:1, took s 2640 // // 30 / 39

31 wordcount.py from pyspark import SparkContext sc = SparkContext(appName="WordCount") // RDD text = sc.textfile("apple.txt") // tab 3 counts = text.flatmap(lambda line: line.split( \t )[2]. // split( ) if len(line.split( \t )) == 4 else )\ // (,1).map(lambda word: (word,1))\ // (,a) (,b) (, a+b).reducebykey(lambda a,b: a+b) 31 / 39

32 wordcount.py // (, x ) (x, ) output = counts.map(lambda x:(x[1],x[0])).\ // Key 10 sortbykey(ascending=false).take(10) // for (count, word) in output: print("%s: %i" % (word.encode( utf_8 ),count)) // Spark sc.stop() / 39

33 // Twitter % hadoop fs -put Apple.txt // % spark-submit wordcount.py // RT: Apple: the: to: : : // RT Apple 33 / 39

34 Spark Transformations RDD RDD Transformations Actions Action 34 / 39

35 Transformations filter: map: flatmap: reducebykey: sample: 35 / 39

36 Action collect: count: take: k 36 / 39

37 RDD join: subtract: 37 / 39

38 Twitter NAIST iphone6s? PS4, XBox Good? PS4 XBox? 38 / 39

39 Apple (E) GalaxyS6 (E) ipad (E) iphone6/6s (E/J) iphone (E) (J) PS4 (E) SONY (E) Surface (E) XboxOne (E) Xbox (E) Xperia (E) (J) 39 / 39

ビッグデータアナリティクス - 第3回: 分散処理とApache Spark

ビッグデータアナリティクス - 第3回: 分散処理とApache Spark 3 : Apache Spark 2017 10 20 2017 10 20 1 / 32 2011 1.8ZB 2020 35ZB 1ZB = 10 21 = 1,000,000,000,000 GB Word Excel XML CSV JSON text... 2017 10 20 2 / 32 CPU SPECfp Pentium G3420 77.6 8,946 Xeon Gold 6128

Spark と大規模データ処理 - NAISTビッグデータアナリティクス 第2回

Spark と大規模データ処理 - NAISTビッグデータアナリティクス第2回