Hadoop Definitive Guide Byte Sized Notes - 6

Chapter 7 MapReduce Types and Formats

General form of map & reduce 

• map: (k1,v1) -> list(k2,v2)
• reduce: (k2,list(v2)) ->list(k3,v3)
• Keys emitted during the map phase must implement WritableComparable so that the keys can be sorted during the shuffle and sort phase.
• Reduce input must have the same types as the map output
• The number of map tasks are not set, the number of map tasks are equal to the number of splits that the input is turned 

Input formats 

• Data is passed to the mapper by InputFormat, Input Format is a factory of RecordReader object to extract the (key, value) pairs from input source
• Each mapper deals with single input split no matter the file size
• Input Split is the chunk of the input processed by a single map, represented by InputSplit
• InputSplits created by InputFormat, FileInputFormat default for file types 
• Can process Text, XML, Binary 

RecordReader 

• Ensure key, value pair is processed
• Ensure (k, v) not processed more than once
• Handle (k,v) which get split

Output formats: Text output, binary output 

Chapter 8 MapReduce Features

Counters

• Useful for QC, problem diagnostics
• Two counter groups task & job
• Task Counter
• gather information about task level statistics over execution time
• counts info like, input records, skipped records, output bytes etc
• they are sent across the network
• Job Counter
• measure job level statistics
• User defined counters 
• Counters are defined by a Java enum
• Counters are global

Sorting

• SortOrder controlled by RawComparator
• Partial sort, secondary sort (Lab)

Joins

To run map-side join use CompositeInputFormat from the org.apache.hadoop.mapreduce.joinpackage

Side Data: Read only data required to process main data set, ex skip words for wordcount

Distributed Cache: To save network bandwidth, side data files are normally copied to any particular node once per job.

Map only Jobs: Image Processing, ETL, Select * from table, distcp, I/p data sampling 

Partitioning 

Map/reduce job will contains most of the time more than  1 reducer.  So basically, when a mapper emits a key value pair, it has to be sent to one of the reducers. Which one? The mechanism sending specific key-value pairs to specific reducers is called partitioning (the key-value pairs space is partitioned among the reducers). A Partitioner is responsible to perform the partitioning.

For more info this article has detailed explanation 

http://www.philippeadjiman.com/blog/2009/12/20/hadoop-tutorial-series-issue-2-getting-started-with-customized-partitioning/

Zero reducers only map tasks

• One reducer when the amount of data is small enough to be processed comfortably by one reducer

Combiners

• Used for reducing intermediate data
• Runs locally on single mapper’s  o/p
• Best suited for commutative and associative operations
• Code identical to Reducer 
• conf.setCombinerClass(Reducer.class);
• May or may not run on the output from some or all mappers

Hadoop Definitive Guide Byte Sized Notes - 4

Chapter 5 Map Reduce Process Flow

General dev process

• Write & test map & reduce functions in dev/local
• Deploy job to cluster
• Tune job later

The Configuration API

• Collection of configuration properties and their values are instance of Configuration class (org.apache.hadoop.conf package)
• Properties are stored in xml files
• Default properties are stored in core-dafult.xml, there are multiple conf files for multiple settings
• If mapred.job.tracker is set to local then execution is local job runner mode

Setting up Dev

• Pseudo-distributed cluster is one whose daemons all run on the local machine
• Use MRUnit for testing Map/Reduce
• Write separate classes for cleaning the data 
• Tool, toolrunner helpful for debugging, parsing CLI

Running on Cluster

• Job's classes must be packaged to a jar file
• setJarByClass() tells hadoop to find the jar file for your program
• Use Ant or Maven to create jar file
• On a cluster map and reduce tasks run in separate JVM
• waitForCompletion() launches the job and polls for progress
• Ex: 275GB of input data read from 34GB of compressed files broken into 101Gzip files
• Typical Job id format: job_YYYYMMDDTTmm_XXXX
• Tasks belong to Jobs, typical TaskID format task_Jobid_m_XXXXXX; m->;map, r->;reduce
• Use MR WebUI to track Jobtracker page/ tasks page/nematode

Debugging Job

• Use WebUI to analyse tasks & tasks detail page 
• Use custom counters
• Hadoop Logs: Common logs for system daemons, HDFS audit, MapRed Job hist, MapRed task hist

Tuning Jobs

Can I make it run faster? Analysis done after job is run

• If mappers are taking less time, then reduce mappers, let each one run longer
• # of reducers should be less than reduce slots 
• Combiners associate properties, reduce map output
• Map output compression
• Serialization other than Writeable
• Shuffle tweaks
• Profiling Tasks: HPROF profiling tool comes with JDK

More than one MapReduce & workflow management

• ChainMapper/ChainReducer to run multiple maps
• Linear chain job flow 
• JobClient.runJob(conf1);
• JobClient.runJob(conf2);
• Apache Oozie used for running dependent jobs, composed of workflow engine, coordinator engine (Lab)
• unlike JobControl Oozie runs as a service 
• workflow is a DAG of action node and control flow nodes
• Action node moving files in HDFS, MR, Pig, Hive
• Control Flow coordinates actions nodes
• Workflows written in XML and has three control-flow nodes and one action node: a start control node, a map-reduce action node, a kill control node, and an end control node. 
• All workflows must have one start and one end node.
• Packaging, deploying and running oozie workflow job
• Triggered by predicates usual time interval or date or external event 

Hadoop Definitive Guide Byte Sized Notes - 3

Chapter 4 Hadoop IO

 

Data Integrity in Hadoop

• For data integrity usually CRC-32 is used but with low end hardware it could be checksum which is corrupt not the data
• Datanodes are responsible for verifying the data they receive before storing the data and its checksum
• Applies to data that they receive from clients and from other datanodes during replication
• Each datanode keeps a persistent log of checksum verifications

LocalFileSystem

• Hadoop LocalFileSystem performs client-side check summing
• Creates hidden .filename.crc
• To disable checksum use RawLocalFileSystem
• FileSystem fs = new RawLocalFileSystem();

Compression

• More space to store files, faster data transfer across network
• Splittable shows whether the compression format supports splitting, that is, whether you can seek to any point in the stream and start reading from some point further on
• Compression algo have space/time trade off
• Gzip sits in middle
• BZip compression better than Gzip but slow
• LZO, LZ4,Snappy are faster
• BZip is only splittable

Codecs

• A codec is the implementation of a compression-decompression algorithm
• We can compress file before writing, for reading a compressed file, CompressionCodecFactory is used to find file name extension and appropriate algo is used

Compression and Input Splits

• Use container file format such as Sequence, RCFile or Avro
• Use compression algo which support splitting

Using Compression in MapReduce

• To compress the output of a MapReduce job, in the job configuration, set the mapred.output.compress property to true
• We can compress output of a map

Serialization

• Serialization process of converting objects to byte stream for transmitting over network, De-serialization is the opposite
• In Hadoop, inter-process communication between nodes is implemented, using remote procedure calls(RPCs), RPC serialization should be compact, fast, extensible, and interoperable
• Hadoop uses its own serialization format, Writables, Avro is used to overcome limitation of Writables interface
• The Writable Interface 

            Interface for serialization

            IntWritable wrapper for Java int, use set() to assign value

            IntWritable writable = new IntWritable();

            writable.set(163);

            or 

            IntWritable writable = new IntWritable(163);

• For writing we use DataOutput binary stream and for reading we use DataInput binary stream
• DataOutput convert 163 into byte array, to de serialize the byte array is converted back to IntWritable
• WritableComparable and comparators
• Comparison of types is crucial for MapReduce
• IntWritable implements the WritableComparable interface
• This interface permits users to compare records read from a stream without de serializing them into objects (no overhead of object creation)
• keys must be WritableComparable because they are passed to reducer in sorted order
• There are wrappers for most data types in Java, boolean, byte, short, int, float, long, double, text is UTF-8 sequence
• BytesWritable is a wrapper for an array of binary data
• NullWritableis a special type of Writable, as it has a zero-length serialization. No bytes are written to or read from the stream
• ObjectWritable is a general-purpose wrapper for the following: Java primitives, String, enum, Writable, null, or arrays of any of these type
• Writable collections
• There are six Writablecollection types, ex for 2D array use TwoDArrayWritable
• We can create custom writable class and it can implements WritableComparable (Lab)

Serialization Frameworks

• Hadoop has an API for pluggable serialization frameworks

Avro

• Language-neutral data serialization system

Why Avro?

• Written in JSON, encoded in binary
• Language-independent schema has rich schema resolution capabilities
• .avsc is the conventional extension for an Avro schema, .avro is data
• Avro’s language interoperability

File-Based Data Structures

• Special data structure to hold data

SequenceFile

• Binary encoded key, value pairs
• To save log files text is not suitable
• Useful for storing binary key value pairs, .seq ext
• SequenceFileInputFormat Binary file (k,v)
• SequenceFileAsTextInputFormat (key.toString(), value.toString())

MapFile

• Sorted SequenceFilewith an index to permit lookups by key
• MapFile.Writer to write, .map ext
• Numbers.map/index contains keys
• Variation of mapfile
• SetFile for storing Writable keys
• ArrayFile int index
• BloomMapFile fast get()
• Fix() method is usually used for recreating corrupted indexes 
• Sequence files can be converted to mapfiles 

Hadoop Definitive Guide Byte Sized Notes - 2

Chapter 3 HDFS

• Hadoop can be integrated with local file system, S3 etc
• Writes always happen at the end of the file, cannot append  data in between
• Default block size 64 MB

Name Node

• Name node holds the file system metadata, namespace and tree  in memory, without this metadata there is no way to access files in cluster 
• Stores this info on local disk in 2 files name space image & edit log
• NN keeps track of which blocks makes up a file
• Block locations are dynamically updates
• NN must be running at all times
• When clients reads a file NN will not be a bottleneck

Secondary Name Node 

• Stores the partial of name node’s namespace image and edit log
• In case of name node failure we can merge namespace image and edit log 

HDFS Federation

• Clusters can scale by adding name nodes, each NN manages a portion of file system namespace 
• Namespace volumes are independent, Name Nodes don’t communicate with each other and doesn’t affect the availability of other name nodes

High Availability 

• Pair of NN in active-stand by config
• NN’s share use shared storage to share edit log
• DN send block reports for both the NN

Fail Over

• NN runs a process to monitor failures, if failover detected then it gracefully terminates

Fencing

• Previously active name node is prevented from doing damage following will help prevent corruption of active name node
• Disable the network port
• Kill all process
• Revoke access to shared storage

File System

• Cannot execute a file in HDFS
• Replication not applicable to directories 
• Globbing wild cards to match multiple files in a directory, sample file selection: input/ncdc/all/190{1,2}.gz


Reading Data 

• FileSystemAPI contains static factory methods to open i/p stream
• A file in HDFS is path object
• Default file system properties are stored in conf/core-site.xml  
• FSDataInputStream
• FileSystem uses FSDataInputStream object to open a file
• positioned readable, seek() expensive

• If FSDataInputStream fails then sends error to NN  and  NN sends the next closet block address

Writing Data

• FSDataOutputStream main class
• If FSDataOutputStream fails then pipeline is closed, partial blocks are deleted, bad data node removed from pipeline, NN replicates under replicated blocks

Replica Placement

• 1st replica at random node, 2nd replica off rack to random node, 3rd replica same rack as 2nd but diff node, applicable to reducers output as well

Coherency 

• Same data across all replicated data nodes, sync()

HAR Hadoop Archives

• Typically used for small data sets
• Packs all small blocks efficiently 
• Archive tool
• Limitations
• No archive compression
• Immutable to add or remove recreate HAR

Hadoop Definitive Guide Byte Sized Notes - 1

Chapter 1

A common way of avoiding data loss is through replication, storage - HDFS, analysis- MapReduce

Disk Drive

• Seek Time: the time taken to move the disk’s head to a particular location on the disk to read or write data, typical RDMBS use B-Tree which has good seek time

• Transfer Rate: streaming of data, disk bandwidth for Big Data apps we need high transfer rate, coz, its write once read many times

New in Hadoop 2.X

• HDFS HA
• HDFS Federation
• YARN MR2
• Secure Authentication

Chapter 2 MapReduce

i/p -> map -> shuffle -> sort -> reduce -> o/p

Some diff between MR1/MR2

1. mapred -> mapreduce
2. Context object unifies JobConf, OutputCollector, Reporter
3. JobClient deprecated
4. map o/p: part-m-nnnn, reduce o/p:  part-r-nnnnn
5. iterator to iterable  

Data Locality

• To run a map task on the data node where the data resides
• map task writes output to local disk not HDFS
• No data locality for reduce, o/p of all mappers gets transferred to reduce node
• # of reduce tasks are not governed by size of i/p
• 0 reduce tasks are possible

Combiner

• performed on the output of the map task
• associate property max(0,20,10,15,25) = max( max(0,20,10), max(25,15))
• • boosts the performance, pre filtering data before sent across the wire and reduce has smaller data set to work with

Big Data Camp 27/10/2013

Big Data Camp was my first un-conference in big data. It’s really different compared to a regular conferences, the sessions are created on the fly, walk in and walk out are welcome. From what I have seen, it was non-technical, focused on the noob to intermediate skill market cap. The host Dave, managed the event very eloquently. Overall, it was good event. I complied small notes on what I liked.   

Talk1: Hadoop in consumer security products

• This organization implements hadoop technology to analyze the sales of their flagship consumer security products
• Existing data warehouse products and tools were OK, although they saw lot of customer’s showing up on their site, the sales reduced.
• The product was not at fault, but the customer analytics was.  The primary touch points or data points viz, server log, click stream etc were fed to hadoop, analytics was performed using sql, hive, mahout, used linear regression for modeling their data.

Talk 2: How to choose data analytics product

This presenter works at Datapad. “There is no one right way, there is a right way for your data, for your team” kinda made sense. Advocated the use of Python and Pandas library.

Talk 3: Block box vs transparent data modeling

I was trying to figure out what his point was, this is my understanding, the functionality of any google product is basically a black box and he was not cool with that or whatever.

Talk 4: Mainframe + Hadoop

This talk was pretty interesting, the presenter works at Syncsort company in CA.  

• 75% of data stored in mainframes (fact)
• Insurance, Retails and Banks store data in mainframes
• Offload mainframe data and batch process it in hadoop
• Mainframe store in ebcdic data type, (new info)
• ADP is hiring people for mainframes seriously?? (Maybe I should learn COBAL.. heck no!!)

Talk 5: Time series database – things happening in time

This was cool too, open source database called InfluxDB - Database based on events,

• HTTP native, show and do the analytics in a browser
• Read/write, manage and security with Http
• I asked the presenter, how is this product different form Storm, Spark & StreamInsight, could not answer my question 100%

Some other random tidbits, big data GUI tools, IBM Big sheets, Datameer &Talend

Preconfigured Single Node Hadoop Distributions

Setting up Hadoop can be quite a hassle if you are not familiar with Linux. Below are a short list of vm’s which are preconfigured with single node Hadoop distribution.  

IBM Big Insight

http://www.ibm.com/developerworks/downloads/im/biginsights/

Pivotal

http://gopivotal.com/pivotal-products/pivotal-data-fabric/pivotal-hd#4

Cloudera  Quick start VM

https://downloads.cloudera.com/demo_vm/virtualbox/cloudera-quickstart-vm-4.3.0-virtualbox.tar.gz

MapR – Choose M3

http://www.mapr.com/products/download/download-mapr-on-premise

Hortonworks

http://hortonworks.com/products/hdp/

Wandisco

http://www.wandisco.com/hadoop/download

and for the brave souls who want to setup Hadoop from scratch (its easy actually)

http://www.michael-noll.com/tutorials/running-hadoop-on-ubuntu-linux-single-node-cluster/

The people & business behind Hadoop

Every decade has a key technology and a wave of new companies breed up around this technology. For the 90’s it was write once run anywhere Java, for the 2000’s the whole world was caught up in the web. For the 2010 decade, the software section is buzzing with the term Big Data along with a significant increase in consumer electronics such as tablets and phones. The key software framework which is most sought out for understanding Big Data is Hadoop. There are three main next gen companies which want customers to ask big questions and brave into new frontiers. In this blog I want to write about Hadoop and its allies.

Hadoop was created by Doug Cutting (used to work for Yahoo, currently Chief Architect at Cloudera & Director at Apache Software Foundation) and Mike Cafarella (currently a professor of computer science at University of Michigan) in 2005

Hadoop was quickly recognized as the go to solution for distributed computing using commodity hardware. This framework promised painless deployment, easy maintenance and was at the right time for handling the big data explosion (being a bit dramatic).  Ever since its release under the Apache federation foundation this project amassed lot of committers; soon an ecosystem was built around Hadoop. Variety of companies started providing Hadoop consulting services, amongst them the most popular are Cloudera, Hortonworks & MapR 

Cloudera

Three top engineers from Google, Yahoo and Facebook (Christophe Bisciglia, Amr Awadallah and Jeff Hammerbacher) foresaw a need for analyzing, managing and processing big data, so they teamed up with an Oracle employee and formed Cloudera in the early 2008.

Open Source Contribution

The employees of this company are active participants to many open source projects, some popular projects are:

Apache Avro, Apache Bigtop, Apache Crunch, Apache Flume, Apache Hadoop, Apache Hive, Apache Lucene, Apache MRUnit, Apache Oozie, Apache Sqoop, Apache Whirr, Apache ZooKeeper, Cloudera Development Kit, Crepo, Hue, Impala, Kitten, ML, Seism, CDH - 100% 
Open Source Hadoop Distribution

Other services provided by this company include a Cloudera Manager & Navigator- Centralized System Management which provide a rich set of features which include

Manager

Deployment & Configuration, Service Management, Service & Host Monitoring, Diagnostics API, Rolling Updates/Restarts, SNMP Support, LDAP Integration, Configuration History & Rollbacks, Operational Reports, Automated Disaster Recovery, BDR Add-on

Navigator

Data Audit – HDFS, HBase & Hive, Navigator Add-on, Access Management, Technical Support and Indemnity, Core Projects, Apache HBase, RTD Add-on, Cloudera Impala, RTQ Add-on, Cloudera Manager, Cloudera Navigator, Navigator Add-on

Partners & valuation

Over 160 active partners

Current net valuation of $700 million at least that the word on the street, along with an additional $65 million through Series E round of venture capital funding by Accel patners .

Developers Community

Cloudera offers training for only 5 days at a price of around $2500. This training includes Hadoop development, administration and data science. Also they provide certification as well.  

Hortonworks

This company was incubated by Yahoo. The employees of this company are regular committers to various Apache Hadoop eco systems projects. Primarily this company provides consulting services and offers an end to end solution with the Hadoop framework. They are emerging out with various data products with the newer version of Hadoop. 

Open Source Contribution

Yarn, Tez, Stinger, Ambari, HDP platform

Partners & Valuation

Microsoft, Teradata, Talend, Rackaspace are official partners. This company is currently valued at $200 million.

Developers Community

Hortonworks offers a similar training like Cloudera for only 5 days at a price of around $2500. This training includes Hadoop development, administration and data science. Also they provide certification as well. The advantage with the Hortonworks is they are committed with Microsoft. They provide training for Hadoop on Windows platform.

MapR

My personal opinion their business model is not viable, although backed up with EMC and Cisco, their online presence is dominated by Cloudera and Hortonworks. This company provides a commercial Hadoop solution by offering three products M3, M5 and M7. The employees of this firm actively participate with HBase, Pig, Apache Hive, ZooKeeper.

Partners & valuation
Google Compute Engine, EMC Cisco, valued at $30 million

What’s in it for us?

Well Hadoop is free, anyone can download it and use it, and the trick is to know what to use it for and how to manage it. There is a need for distributed computing platform and Hadoop by itself is in its nascent stages. It’s definitely not a threat to existing database platforms however investing time and money with Hadoop would be a safe bet for future.