Hadoop – Tips
String cacheFilePath = args[3];
DistributedCache.createSymlink(jobconf);
try {
DistributedCache.addCacheFile(new URI(cacheFilePath), jobconf);
} catch (URISyntaxException e) {
driverLogger.error(“Exception in adding the countrylookup cache file”);
}
jobconf.set(NAC2Constants.iso2_cc_filename, cacheFilePath.substring(cacheFilePath.indexOf(“#”) + 1, cacheFilePath.length()));
——————————————————————————————————————————————————————————http://codingjunkie.net/secondary-sort/
——————————————————————————————————————————————————————————
Install Orange
Install Orange: (http://orange.biolab.si/download/)
Mac OS X:
sudo easy_install pip
pip install https://bitbucket.org/biolab/orange/get/tip.tar.gz#egg=Orange==2.7.2
sudo easy_install numpy
sudo easy_install orange
Help Links:
http://orange.biolab.si/start-using/
http://orange.biolab.si/docs/latest/tutorial/rst/classification/#learners-and-classifiers
AWS – Help Tips
Uploading files to EMR MasterNode:
ssh hadoop@ec2-54-211-65-155.compute-1.amazonaws.com -i ~/documents/AWS_home/cred/sariyaba.pem
Use Cyberduck to SFTP files from Mac to EC2 instance
hadoop jar /home/hadoop/poc/dynamodb-0.0.1-SNAPSHOT.jar com.here.poc.dynamodb.LoadDriverNew s3://ariyabala/POC/DynamoDB/sample s3://ariyabala/POC/DynamoDB/output
Deploying files and Executing the jar:
scp -i ~/documents/AWS_home/cred/sariyaba.pem target/nokia.ddb-0.0.1-SNAPSHOT.jar hadoop@ec2-54-226-87-206.compute-1.amazonaws.com:/home/hadoop/
ssh -i ~/Documents/AWS_home/cred/sariyaba.pem hadoop@ec2-54-211-107-39.compute-1.amazonaws.com
inputPath=”s3://ariyabala/POC/spring/nokia_text.txt” outputPath=”s3://ariyabala/POC/spring/output_wc” hadoop jar nokia.ddb-0.0.1-SNAPSHOT.jar com.nokia.ddb.Job wordCountJob
Install S3 CLI in Mac:
ruby -e “$(curl -fsSL https://raw.github.com/mxcl/homebrew/go)”
brew install s3cmd
Upload/Download Files to/from S3:
s3cmd -c [configuration file] get –recursive s3path localpath
s3cmd -c ~/Documents/s3cfg-struct get –recursive s3://ariyabala/ddb/geoip-export/2013-10-28_07.00/ ~/Documents/AWS_home/POC/DDB/ddb_exports/
s3cmd -c [configuration file] put localpath s3path
s3cmd -c ~/Documents/s3cfg-struct put ~/Documents/workspace/*.txt s3://ariyabala/POC/spring/
Recursive Line count on files in S3:
output=$(path=`hadoop dfs -lsr s3://com.nokia.analytics.prod.deviceact/activation/processed/2012-11-04/ | awk ‘{print $6}’| grep ‘part’ |grep -v NativeS3FileSystem` ;for f in $path; do count=`hadoop dfs -cat s3://com.nokia.analytics.prod.deviceact/$f | wc -l`;echo $f”:”$count”\n”; done;); echo -e $output
Run Script:
./elastic-mapreduce –create –alive –name “My Development Jobflow” –jar s3://elasticmapreduce/libs/script-runner/script-runner.jar —args “s3://<buckname>/test.sh”
Mounting Instance Store:
sudo umount /dev/xvdb
sudo mkfs.ext4 /dev/xvdb
sudo mkdir -p /local/b
sudo mount /dev/xvdb /local/b
sudo umount /dev/xvdc
sudo mkfs.ext4 /dev/xvdc
sudo mkdir -p /local/c
sudo mount /dev/xvdc /local/c
Unit 6 – Connect
A => B, If A is true, then B is True
Logically this is equivalent to ~A U B
Unit 5 – Learn
Clustering:
Group of similar user/user-queries based on terms
group of similar comments based on words
group of animal observation having similar features
Find regions that are more populated than random data
i.e regions where r = P(X)/Po(X) is large ( Po(X) is uniform)
f(X) = E[y|X] = r/(1+r)
Rule Mining: Clustering Features
Association Rule Mining:
Nice to Read: http://www.academia.edu/648890/Support_vs_Confidence_in_Association_Rule_Algorithms
Infer rule A,B,C => D if
high support: P(A,B,C,D) > s
high confidence: P(D|A,B,C) > c
high Interestingness: P(D |A,B,C) / P(D) > i
Problems with Association Rules:
Characterization of classes:
small classes get left out
use decision trees instead of association rules based on mutual information – costly
Learning rules from data:
high support means negative rules cannot catered as it focusses only on association
use “interesting subgroup discovery” instead
Nice to Read: Beyond market baskets: generalizing association rules to correlations”
One Approach to Latent Models:
Matrix A needs to be written as A ≈ XY (i.e) m X n = (m X k )(k X n) [ m words, n documents, k topics]
since X and Y are always smaller, this is almost always an approximation
so we minimize || A – XY ||F – Here F means sum of squares
Subject to all entries being non-negative – NNMF
Unit 4.2 – Iteration and Map-Reduce
Graph Based Problems – Pregel
Map Reduce Evolution – Iterate
Many applications require repeated MR
- iterate MR:
- but make it more efficient – Avoid Data copy (HaLoop, Twister)
- generalized data-flow graph of Map Reduce tasks
- tasks are blocking for Fault Tolerant (Dryad/LINQ, Hyracks)
- direct implementation of recursion in MR
- How to recover from non-blocking tasks failing?
- graph model: (Pregel, Giraph – Hadoop, Graphlab, GPS)
- stream model: (S4)
Repeated Matrix Vector Products using MR:
Matrix Vector Product as Graph operation:
Pregel ( Graph Programming Model):
Each Vertex executes the code once the graph is loaded
Each Vertex knows its outgoing edges
Vertices are distributed across processors
SuperStep: Each Vertex
receives messages from in-neighbours
… computation…
sends messages to out-neighbours
decides whether or not to halt
(halt: vertex stays inactive except if it receives a message)
computation halts if all vertices are in halt state
Pregel Summary:
- Appropriate if problem can be mapped to a Graph
- Data needs to fit in total memory
- avoids reloading data between iterations
- need to properly distribute vertices to machines: graph partitioning
- Important Features:
- Combiners – Avoid repeating identical messages
- aggregation – global checks
- Graph mutation – vertices can modify the graph
Have a look: http://www.cs.rochester.edu/u/sandhya/csc258/seminars/naim_mapreduce.pdf
Unit 4.1 – Introduction to Graph Data
Resource Description Framework Graphs
- Model graphs are a bag of edges
- Every edge is a triple.
- <subject><predicate><object>
- Each node/edge has a unique identifier (URI)
Graph Query Language:
- Datalog
- XPath
- Blueprints/Gremlin
- Rich JDBC-like Graph traversal API
- SPARQL:
- Query by Patterns or template of Patterns
- Proposed extensions to include full graph traversal support
Linked Open Data and Challenges:
- 30 Billion edges (RDF) and growing
- Graph is growing larger
- Recursive Traversals
- Data partitioning is not easy
- Complex Interconnectivity
- Pay as you grow model
Graph Data Management:
- Transactional GDM
- Neo4J
- Jena
- HyperGraphDB
- RDF3x
- Analytical GDM
- Pregel
- Giraph
- Hama
- Both
- Trinity
- Horton
- Titan
Graph Problems:
- Reachability Queries
- Shortest path Queries
- Pattern Queries
HW:
Partitioning RDF graphs for parallel execution is hard because
- Graphs can continuously change
- Interconnected nature does not lend for effective partitioning
- Graph pattern matching queries
- Reachability queries
Unit 3 – Web Intelligence and Big Data – Load
Why Big Data Technology?
Challenges of Traditional Data warehousing Technology:
- Fault Tolerance at scale
- Variety of Data Types
- Manage data volumes without archiving
- Parallelism
Parallel Computing:
Speed up S = T1/Tp, Time with p processors vs with one
Efficiency E = T1/pTp
Scalable Algorithm – E, increasing function of n/p, Where n is the ‘problem size ‘
Parallel Efficiency of Map Reduce:
Post Map Data – σD
P Processors – Mappers+Reducers
wD – Useful word to be done
OverHeads – σD/P – Intermediate data to be written by each Mapper
Time for transmitting to P reducers = (OverHeads/P reducers)* P mapper results = (σD/P2)*P = σD/P
Efficiency of MR – εMR = Useful Work to be done / Work done by MR
Useful Work done = wD
Work done by MR = P ( work done by each process + Intermediate write + Intermediate read work done by each process)
= P ( wd/P + (2c σD/P) )
εMR = wd / P ( wd/P + (2c σD/P) ) = 1 / (1 + 2cσ/w)
This implies Efficiency of MR is independant of P. Therefore it is scalable.
Efficiency approaches 1 if useful work done per data item grows, independent of P.
Calculating Parallel Efficiency for WordCount:
n – No of Documents
m – No of words
f – Average frequency of the words occuring in each document
Total Data = nmf
Intermediate data = atmost m words * P processes = mP
σD = mP
σ = mP/D = mP/nmf = P/nf
Substituting in Efficiency equation
εMR = 1 / (1 + 2cσ/w) = 1 / (1 + 2c P/wnf)
Assuming c and w are same,
εMR = 1/(1 + 2P/nf)
as n/P –> ∞, εMR ==> 1
MR Code
Mapred Package:
Driver Class:
package com.here.poc.dynamodb;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class LoadDriver extends Configured implements Tool {
static RunningJob job;
public int run(String[] args) throws Exception {
// TODO Auto-generated method stub
Configuration conf = getConf();
JobConf jobConf = new JobConf(conf, LoadDriver.class);
jobConf.setJobName(“Bulk Load into Dynamo DB”);
jobConf.setMapperClass(LoadMapper.class);
FileInputFormat.addInputPath(jobConf, new Path(args[0]));
FileOutputFormat.setOutputPath(jobConf, new Path(args[1]));
job = JobClient.runJob(jobConf);
return 0;
}
public static void main(String[] args) throws Exception {
// TODO Auto-generated method stub
int exitCode = ToolRunner.run(new LoadDriver(), args);
long total = job.getCounters().findCounter(Counters.Total).getValue();
long inserted = job.getCounters().findCounter(Counters.Inserted).getValue();
long error = job.getCounters().findCounter(Counters.Error).getValue();
System.out.println(exitCode);
}
}
Mapper Class:
package com.here.poc.dynamodb;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
@SuppressWarnings(“deprecation”)
public class LoadMapper extends MapReduceBase implements Mapper<LongWritable, Text, LongWritable, Text> {
public void map(LongWritable arg0, Text arg1,
OutputCollector<LongWritable, Text> arg2, Reporter arg3)
throws IOException {
// TODO Auto-generated method stub
}
@Override
public void close() throws IOException {
// TODO Auto-generated method stub
super.close();
}
@Override
public void configure(JobConf job) {
// TODO Auto-generated method stub
super.configure(job);
}
}
Mapreduce:
Driver Class:
import java.io.File;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.here.poc.dynamodb.DBUtils.DynamoDBAction;
public class LoadDriverNew extends Configured implements Tool {
static Job job;
int retCode;
AmazonDynamoDBClient dynamoDB;
DynamoDBAction dbAction;
public int run(String[] args) throws Exception {
// TODO Auto-generated method stub
job = new Job(new Configuration(),”Load DDB New”);
Configuration conf = job.getConfiguration();
job.setJarByClass(LoadDriverNew.class);
job.setNumReduceTasks(0);
System.out.println(args.length);
if(args.length == 2){
job.setMapperClass(LoadMapperNew.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
retCode = job.waitForCompletion(true)?0:1;
}else{
retCode=-1;
}
return retCode;
}
public static void main(String[] args) throws Exception {
// TODO Auto-generated method stub
int retCode = ToolRunner.run(new LoadDriverNew(), args);
System.exit(retCode);
}
}
Mapper Class:
mport java.io.File;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import com.amazonaws.auth.PropertiesCredentials;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.here.poc.dynamodb.DBUtils.DynamoDBAction;
public class LoadMapperNew extends Mapper<LongWritable, Text, Text, NullWritable> {
AmazonDynamoDBClient dynamoDB;
DynamoDBAction dbAction;
@Override
protected void setup(Context context) throws IOException,
InterruptedException {
// TODO Auto-generated method stub
super.setup(context);
}
@Override
protected void cleanup(Context context) throws IOException,
InterruptedException {
// TODO Auto-generated method stub
super.cleanup(context);
}
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
context.write(new Text(“”), NullWritable.get());
}
}
Unit 2 – Web Intelligence and Big Data – Listen
Preamble:
Search Engine, Social Networking site, Auction sites live on ads. Thereby they need to
- Recognize and Distinguish Shopper and Surfer from a browser
- Gauge opinion and sentiment
- Understand what people say
- Machine learning from Information Theory
Shannon Theory – Measuring Information:
Information is related to Surprise.
Dog bites a man – Not news
Man bites a dog – Interesting !!!!!!
Message informing us of an event that has the probability conveys
– log 2 p gives the bits of information
TF – IDF:
Rarer words make better keywords
IDF = Inverse document frequency of word w = log 2 (N/Nw)
where
N = Total documents
Nw = Documents containing word w
More frequent words make better keywords
(nw)d = Frequency of w in document d
TF – IDF = term-frequency x IDF = (nw)d log 2 (N/Nw)
Document that has the word Turing 15 times is more likely about Computer Science than one with 2 Occurrences
TF – IDF was invented as a heuristic technique
Mutual information between all words and all pages is proportional to
Σd Σw (nw)d log 2 (N/Nw)
Language and Statistics:
Does similar documents have similar TF-IDF keywords?!!
Semantics of a word-use depend on context. Is it computable?!!
Is semantics i.e Meaning just Statistics ?!! or more
Machine learning : Shoppers or Surfers:
Machine learning is all about learning from the past data.
– past behaviour of many many searchers using these keywords
For ex: With Keywords Red, Flower, Gift, Cheap what is the likely chances that the user will end up in a buy
Prediction Using Conditional Probability:
Determine P(B) , given R, F , G, C. In other words P(B|R, F, G, C)
Sets, Frequencies and Bayes Rule:
p(B|R) = (i/r)
p(R) = (r/n)
p( B and R ) = i/n = i/r * r/n = p(B|R) p(R)
This is Bayes Rule.
p(B,R) = p(B|R) p(R) = p(R|B) p(B) = i/k * k/n
Naïve Bayer’s Classifier:
P(B|R,C) P(R,C) = P(R,C|B) P(B) – Bayer’s rule
= P(R|C,B) P(C|B) P(B) – Bayer’s rule
= P(R|B) P(C|B) P(B) – R and C are independent
Given values r and c for R and C
compute:
p(r|B=y) * p(c|B=y) * p(B=y)
p(r|B=n) * p(c|B=n) * p(B=n)
Choose B =y if this is >α(usually 1) and B=n otherwise
Likelihood Ratio:
L = i=1∏ n p(xi |B =y) * p(B=y)
p(xi |B =n) * p(B=n)
Sentiment Vs Machine Learning:
Millions of tweets per day. Companies can listen to the voice of their customers like never before
Sentiment – Brands/ Competitive position … +/- counts
Using the likelihood ratio to define if the new tweet is likely positive or not.
Machine Learning and Mutual Information:
Mutual information between Feature (F) viz Red, Cheap.. and Behaviour (B) viz Buyer, Surfer… is
I(F,B) = f,b Σ p(f,b) log[ p(f,b) / p(f)p(b)
If f and b are independent, then p(f,b) = p(f) p(b), thereby I(F,B) = 0
This implies if F and B are independent there will be no mutual information across both.
Opinion Mining Vs Sentiment:
Using Information Theory we can figure whether or not it is positive or negative comment/compliant.
But what are they complaining about ?!!!!
Example:
Book me an American flight to New york. I hate British food.
– Here the user may probably refer the word American to airliner rather than the nationality
Book me an American flight to New york. I hate English food.
– Here the user may probably refer the word American to Nationality rather than the airliner
Took the car on a terrible, bumpy road,it did well enough.
– Is the family happy with their new car?
– Does Bayer classifier work good in this? probably not !!!!!!
Bayesian learning using “bag of words” – is it enough ?!!