WIXLIB

TI2736-BBig Data ProcessingClaudia Hauffti2736b-ewi@tudelft.nl

PatternHadoopCtd.GiraphZooKeeperSparkSpark

Learning objectives Exploit Hadoop’s Counters and setup/cleanupefficiently Explain how Hadoop addresses the problem ofjob scheduling Explain Hadoop’s shuffle & sort phase and usethat knowledge to improve your Hadoop code Implement strategies for efficient data input3

Hadoop ProgrammingRevisited: setup andcleanup4

Setup & cleanup One MAPPER object for each map task Associated with a sequence of key/value pairs (the “input split”) map() is called for each key/value pair by the executionframeworkOne REDUCER object for each reduce task Programmer “hints”the number of mappersto usereduce() is called once per intermediate keyProgrammer canset the number ofreducersMAPPER/REDUCER are Java objects - allows side effects Preserving state across multiple inputs Initialise additional resourcesEmit (intermediate) key/value pairs in one go 5

Setup useful for one-off operations: opening an SQL connection loading a dictionary etc.SetupWordCount* - count only valid dictionary terms1 public class MyMapper extends2Mapper Text, IntWritable, Text, IntWritable {34 private Set String dictionary;//all valid words56 public void setup(Context context) throws IOException {7dictionary Sets.newHashSet();8loadDictionary();//defined elsewhere, reads file from HDFS9 }1011 public void map(Text key, IntWritable val, Context context)12throws IOException, InterruptedException ;15context.write(key, new IntWritable(1));16 }17 }6

Setup useful for one-off operations: opening an SQL connection loading a dictionary etc.SetupWordCount* - count only valid dictionary terms1 public class MyMapper extends2Mapper Text, IntWritable, Text, IntWritable {34 private Set String dictionary;//all valid words56 public void setup(Context context) throws IOException {Called once in the life cycle7dictionary Sets.newHashSet();of a Mapper object: before8loadDictionary();//defined elsewhere, reads file from HDFSany calls to map()9 }1011 public void map(Text key, IntWritable val, Context context)Called once for each key/{12throws IOException, InterruptedExceptionvalue pair that appears in13if(!dictionary.contains(key.toString())the input split14return;15context.write(key, new IntWritable(1));16 }17 }7

single reducer settingCleanupWordCount** - how many wordsstart with the same letter?1 public class MyReducer extends2Reducer PairOfIntString, FloatWritable, NullWritable, Text {3 private Map Character, Integer cache;45 public void setup(Context context) throws IOException {6cache Maps.newHashMap();7 }8 public void reduce(PairOfIntString key, Iterable IntWritable 9values, Context context) throws10IOException, InterruptedException {11char c key.toString().charAt(0);12for(IntWritable iw : values){13//add iw to the current value of key c in cache14}15 }17 public void cleanup(Context context) throws IOException,18InterruptedException {19for (Character c : cache.keySet()) {20context.write(new Text(c), new IntWritable(cache.get(c));21}22 }23 }8

single reducer settingCleanupWordCount** - how many wordsstart with the same letter?1 public class MyReducer extends2Reducer PairOfIntString, FloatWritable, NullWritable, Text {3 private Map Character, Integer cache;4Called once in the life cycle of a5 public void setup(Context context) throwsIOException{Reducerobject: beforeany6cache Maps.newHashMap();calls to reduce()7 }8 public void reduce(PairOfIntString key, Iterable IntWritable 9values, Context context) throws10IOException, InterruptedException {Called once for each key that11char c key.toString().charAt(0);was assigned to the reducer12for(IntWritable iw : values){13//add iw to the current value of key c in cache14}15 }17 public void cleanup(Context context) throws IOException,18InterruptedException {Called once in the life cycle of a19for (Character c : cache.keySet()) {object: after all calls to20context.write(new Text(c), new ReducerIntWritable(cache.get(c));reduce()21}22 }23 }9

Hadoop ProgrammingRevisited: Counters10

Counter basics Gathering data about the data we are analysing, e.g. Number of key/value pairs processed in map Number of empty lines/invalid lines Wanted: Easy to collect Estimates are viewable during job execution (e.g. to stop aHadoop job early at too many invalid key/value pairs) Why not use log messages instead? Write to the error log when an invalid line occurs Hadoop’s logs are huge, you need to know where to look Aggregating stats from the logs requires another pass over it11

Counter basics Gathering data about the data we are analysing, e.g. Number of key/value pairs processed in map Number of empty lines/invalid lines Wanted: Easy to collect Viewable during job execution (stop Hadoop job early at toomany invalid key/value pairs) What about log messages? Write to the error log when an invalid line occurs Hadoop’s logs are huge, you need to know where to look Aggregating stats from the logs requires another pass over it12

Counter basics Counters: Hadoop’s way of aggregating statistics Counters count (increment) Built-in counters maintain metrics of the job MapReduce counters (e.g. #skipped records by allmaps) File system counters (e.g. #bytes read from HDFS)Job counters (e.g. #launched map tasks)You have already seen them13

Counter basics Counters: Hadoop’s way of aggregating statistics Counters count (increment) Built-in counters maintain metrics of the job MapReduce counters (e.g. #skipped records by allmaps) File system counters (e.g. #bytes read from HDFS)Job counters (e.g. #launched map tasks)You have already seen them14

Counter basics Counters: Hadoop’s way of aggregating statistics CountersMap-ReducecountFramework(increment)Map input records 5903Mapcountersoutput records 47102 Built-inmaintain metrics of the jobCombine input records 47102 MapReducecounters(e.g. #skipped records by allCombine outputrecords 8380Reduce output records 5934maps)File System Counters File system counters (e.g. #bytes read from HDFS)FILE: Number of bytes read 118124FILE:Numberof#launchedbytes written 1075029 Jobcounters(e.g.map tasks)HDFS: Number of bytes read 996209HDFS: Number of bytes written 59194 You have already seen them15

Built-in vs. user-defined Built-in counters: exist for each Hadoop job User-defined Counters are maintained by theapplication they are associated with Periodically sent to the Tasktracker and then the(pre-YARN setup)Jobtracker for global aggregation Aggregated per job by the ResourceManager (YARN)Counter values are only definite once the job has completed!Counters may go down if a task fails!16

Code exampleWordCount* - count words and chars123456789101112131415161718enum Records {several enum’s possible:WORDS, CHARS;};used to group counterspublic class WordCount {public static class MyMapper extendsMapper LongWritable, Text, Text, IntWritable {public void map(LongWritable key, Text value,Context context) throws IOException {String[] tokens value.toString().split(" ");for (String s : tokens) {context.write(new Text(s), new rement(s.length());}}}17user-defined counters appearautomatically in the final status output

Code exampleWordCount* - count words and chars123456789101112131415161718enum Records {WORDS, CHARS;several enum’sFrameworkpossible:Map-Reduce};used to group counterspublic class WordCount {Map input records 5903public static class MyMapperextendsrecords 47102Map outputMapper LongWritable, Text, Text, IntWritable {Combine input records 47102Combine key,outputpublic void map(LongWritableText records 8380value,ContextIOException {Reducecontext)output throwsrecords 5934String[] tokens value.toString().split(" "); for (String Recordss : tokens) {context.write(newText(s), new IntWritable(1));CHARS 1);WORDS .length());}}}18user-defined counters appearautomatically in the final status output

Code example II1 enum Records { MAP WORDS, REDUCE WORDS; };23 public class WordCount {4-- MAPPER5public void map(LongWritable key, Text value,6Context context)7throws IOException { 89String[] tokens value.toString().split("");Records10for (String s : tokens) {MAP WORDS 4710211context.write(new Text(s), new IntWritable(1));12context.getCounter(Records.MAP WORDS).increment(1);REDUCE WORDS 4710213}14}15-- REDUCER (Combiner is a copy of the Reducer)16public void reduce(Text key, Iterator IntWritable values,17Context context) throws IOException {18int sum 0;19while (values.hasNext())20sum EDUCE WORDS).increment(sum);22}23 }19

Job Scheduling20

Last time GFS/HDFSdistributed file system: file systems thatmanage the storage across a network ofmachines.21Image source: .google.com/en//archive/gfs-sosp2003.pdf

What about the jobs? Hadoop job: unit of work to be performed Input dataMapReduce programConfiguration informationHadoop divides input data into fixed size input splits One map task per splitOne map function call for each record in the splitSplits are processed in parallel (if enoughDataNodes exist)22

“MapReduce1JobTracker and TaskTracker23Image source: http://lintool.github.io/MapReduceAlgorithms/”

“MapReduce1Hadoop in practice: Yahoo!(2010) ”40 nodes/rack sharing one IP switch16GB RAM per cluster node, 1-gigabit Ethernet70% of disk space allocated to HDFS Remainder: operating system, data emitted byMappers (not in HDFS)NameNode: up to 64GB RAMTotal storage: 9.8PB - 3.3PB net storage (replication: 3)60 million files, 63 million blocks54,000 blocks hosted per DataNode1-2 nodes lost per dayTime for cluster to re-replicateHDFS cluster with 3,500 nodeslost blocks: 2 minutes24

YARN (MapReduce 2) JobTracker/TaskTrackers setup becomes a bottleneckin clusters with thousands of nodes As answer YARN has been developed (Yet AnotherResource Negotiator) YARN splits the JobTracker’s tasks (job scheduling andtask progress monitoring) into two daemons: Resource manager (RM) Application master (negotiates with RM for clusterresources; each Hadoop job has a dedicated master)25

Job scheduling Thousands of tasks may make up one job Number of tasks can exceed number of tasks that canrun concurrently Scheduler maintains task queue and tracks progressof running tasks Waiting tasks are assigned nodes as they becomeavailable “Move code to data” Scheduler starts tasks on node that holds a particularblock of data needed by the task if possible26

Job schedulingFIFO schedulerPriority schedulerFair schedulerCapacity scheduler27

Basic schedulers Early on: FIFO scheduler Job occupies the whole cluster while the restwaits Not feasible in larger clusters Improvement: different job priorities VERY HIGH,HIGH, NORMAL, LOW, or VERY LOW Next job is the one with the highest priorityNo pre-emption: if a low priority job is occupyingthe cluster, the high priority job still has to wait28

Fair Scheduler I Goal: every user receives a fair share of the clustercapacity over time If a single job runs, it uses the entire cluster As more jobs are submitted, free task slots are givenaway such that each user receives a “fair share” Short jobs complete in reasonable time, long jobskeep progressingA user who submits more jobs than a second user willnot get more cluster resources on average29

Fair Scheduler II Jobs are placed in pools, default: one pool peruser Pre-emption: if a pool has not received its fairshare for a period of time, the scheduler will killtasks in pools running over capacity to give moreslots to the pool running under capacity Task kill ! Job kill Scheduler needs to keep track of all users,resources used30

Capacity Scheduler Cluster is made up of a number of queues (similarto the Fair Scheduler pools) Each queue has an allocated capacity Within each queue, jobs are scheduled using FIFOwith priorities Idea: users (defined using queues) simulate aseparate MapReduce cluster with FIFOscheduling for each user31