Yesterday Yahoo announced the Learning to Rank Challenge – a pretty interesting challenge (as the somewhat similar Netflix Prize Challenge also was).

Data and Problem
The data sets contains (to my interpretation) per line:

1. url – implicitly encoded as line number in the data set file
2. relevance – low number=high relevance and vice versa
3. query – represented as an id
4. features – up to several hundreds

and the problem is to find a function that gives relevance numbers per url per query id.

Initial Observation
In dataset 1 there are ~473k URLs and ~19k queries. At first I thought this meant that there are in average 473/19 ~ 24 relevance numbers for each query (see actual distribution of counts in figure below), i.e. corresponding to search result 1 to 24, but it seems like there are several URLs per unique query that has the same relevance (e.g. URLx and URLy both can have relevance 2 for queryZ). The paper Learning to Rank with Ties seems potentially relevant to deal with this.

Multiple URLs that shares relevance for a unique query can perhaps be due to:

1. similar/duplicate content between the URLs?
2. a frequent query (due to sampling of examples?)
3. uncertainty about which URL to select for particular a relevance and query?
4. there is a tie, i.e. they are equally relevant

Potential classification approach?
From a classification perspective there are several (perhaps naive?) approaches that could be tried out:

1. Use relevance levels as classes (nominal regression) and use a multiclass-classifier
2. Train classifier as binary competition within query, i.e. relevance 1 against 2, 3, .., and relevance n against n+1, .. (probably get some sparsity problems due to this)
3. Binary competition across queries, but is problematic due to that a relevance of 4 for one query could be more relevant than a relevance of 1 for a another query (and there is no easy way to determine that directly from the data), but if the observation related to multiple URLs per relevance level per query (see above) is caused by uncertainty one could perhaps use 1/(number of URLs per relevance level per query) as a weight to either:
   1. support training across queries, e.g. a URL for a query with relevance 1 is better that another query of relevance 1 with 37 URLs of that relevance, this approach could perhaps be used somehow using regression? The problem is to compare against different relevance levels, e.g. is a relevance 2 for a query with 1 url more confident than one of relevance 1 for a query with 37 URLs?
   2. use a classifier that supports weighing examples and the approach in 1 or 2.

Conclusion
Still have more questions than answers, so next step is the learning to rank bibliography.

Bookmark to:

Hadoop is a set of open source technologies that supports reliable and cost-efficient ways of dealing with large amounts of data. Given the vast amounts of business critical and required data companies gather (e.g. required due to Sarbanes–Oxley (SOX) or EU Data Retention Directive), Hadoop becomes increasingly relevant.

Hadoop Technologies

Several Hadoop technologies are inspired by Google’s infrastructure.

1. Processing and Storage

1.1 Processing – Mapreduce
Mapreduce can be used to process and extract knowledge from arbitrary amounts of data, e.g. web data, measurement data or financial transactions – Visa reduced their processing time for transactional statistics from 1 month to 13 minutes with Hadoop. In order to use Mapreduce developers need to parallelize their problem and program against an API – here for an example of machine learning with Hadoop. Hadoop’s Mapreduce is inspired by the paper MapReduce: Simplified Data Processing on Large Clusters.

1.2 File Storage – HDFS
HDFS is scalable and distributed file system. It supports configurable degree of replication for reliable storage even when running on cheap hardware. HDFS is inspired by the paper The Google File System

1.3 Database – HBase
HBase is a distributed database that supports storing billions of rows with millions of columns that runs on top of HDFS. HBase can replace traditional databases if they get problems scaling or become to expensive licence-wise, see this presentation about Hbase. HBase is inspired by the paper Bigtable: A Distributed Storage System for Structured Data

2. Data Analysis

Mapreduce can be used to analyze all kinds of data (e.g. text, multimedia, numerical data) and have high flexibility, but for more structured data the following Hadoop Technologies can be used:

2.1 Pig
SQL-like language/system running on top of Mapreduce. Pig is developed by Yahoo and inspired by the paper Interpreting the Data: Parallel Analysis with Sawzall

2.2 Hive
Datawarehouse running on top of Hadoop, developed by Facebook. Query language is very similar to SQL.

3. Distributed Systems Development

3.1 Avro
Avro is used for efficient serialization of data and communication between services. It is in several ways similar to Google’s protocolbuffers and Facebook’s Thrift.

3.2 Zookeeper
Coordination between distributed processes. It is inspired by the paper The Chubby lock service for loosely-coupled distributed systems

3.3 Chukwa
Monitoring of distributed systems.

Bookmark to:

This posting is an update to the similar posting from October 2009, roughly doubling the numbers of papers from the previous posting, the new ones are marked with *

Motivation
Learn from academic literature about how the mapreduce parallel model and hadoop implementation is used to solve algorithmic problems.

Which areas do the papers cover?

Bioinformatics/Medical Informatics
* MapReduce-Based Pattern Finding Algorithm Applied in Motif Detection for Prescription Compatibility Network (2009)
* MrsRF: an efficient MapReduce algorithm for analyzing large collections of evolutionary trees

Machine Translation
Grammar based statistical MT on Hadoop (2009)
Large Language Models in Machine Translation (2008)

Spatial Data Processing
* Experiences on Processing Spatial Data with MapReduce

Information Extraction and Text Processing
* Data-intensive text processing with MapReduce
* Web-Scale Distributional Similarity and Entity Set Expansion (2009)
The infinite HMM for unsupervised PoS tagging (2009)

Artificial Intelligence/Machine Learning/Data Mining
* Residual Splash for Optimally Parallelizing Belief Propagation
* Stochastic gradient boosted distributed decision trees
* Distributed Algorithms for Topic Models
* When Huge is Routine: Scaling Genetic Algorithms and Estimation of Distribution Algorithms via Data-Intensive Computing
* Cloud Computing Boosts Business Intelligence of Telecommunication Industry
* Parallel K-Means Clustering Based on MapReduce
* Large-scale multimedia semantic concept modeling using robust subspace bagging and MapReduce
* Parallel algorithms for mining large-scale rich-media data
* Scaling Simple and Compact Genetic Algorithms using MapReduce
* Scalable Distributed Reasoning using Mapreduce
Scaling Up Classifiers to Cloud Computers (2008)

For an example of Parallel Machine Learning with Hadoop/Mapreduce, check out our previous blog post.

Ads Analysis
Large-Scale Behavioral Targeting (2009)
Search Advertising using Web Relevance Feedback (2008)
Predicting Ads’ ClickThrough Rate with Decision Rules (2008)

Search Query Analysis
BBM: Bayesian Browsing Model from Petabyte-scale Data (2009)
AIDE: Ad-hoc Intents Detection Engine over Query Logs (2009)

Information Retrieval (Search)
* Efficient Clustering of Web Derived Data Sets
* The PageRank algorithm and application on searching of academic papers
* A Parallel Algorithm for Finding Related Pages in the Web by Using Segmented Link Structures
On Single-Pass Indexing with MapReduce (2009)
A Data Parallel Algorithm for XML DOM Parsing (2009)
Semantic Sitemaps: Efficient and Flexible Access to Datasets on the Semantic Web (2008)

Spam & Malware Detection
Characterizing Botnets from Email Spam Records (2008)
- Clustering of emails into spam campaign
- Finding probability that 2 spam messages are sent form same machine
- Estime likelihood of botnets based on common senders in spam campaigns
The Ghost In The Browser Analysis of Web-based Malware (2007)

Image and Video Processing
MapReduce Optimization Using Regulated Dynamic Prioritization (2009)
- Video Stream Re-Rendering
Map-Reduce Meets Wider Varieties of Applications (2008)
- Location detection in images

Networking
Reducible Complexity in DNS

Simulation
Map-Reduce Meets Wider Varieties of Applications (2008)
- Simulation of earthquakes (geology)

Statistics
* Brute Force and Indexed Approaches to Pairwise Document Similarity Comparisons with MapReduce (2009)
Fast Parallel Outlier Detection for Categorical Datasets using Mapreduce (2009)
MapReduce Optimization Using Regulated Dynamic Prioritization (2009)
- Digg.com story recommendations
Calculating the Jaccard Similarity Coefficient with Map Reduce for Entity Pairs in Wikipedia (2008)
- Measuring Wikipedia Editor similarity
Map-Reduce Meets Wider Varieties of Applications (2008)
- Netflix video recommendation
Large-scale Parallel Collaborative Filtering for the Netflix Prize (2008)

Numerical Mathematics
* Mapreduce for Integer Factorization

Graphs
* Distributed Algorithm for Computing Formal Concepts Using Map-Reduce Framework
* Storage and Retrieval of Large RDF Graph Using Hadoop and MapReduce
* Graph Twiddling in a MapReduce World
DOULION: Counting Triangles in Massive Graphs with a Coin (2009)
Fast counting of triangles in real-world networks: proofs, algorithms and observations (2008)

Who wrote the above papers? (section added 20100307)
Companies: China Mobile, eBay, Google, Hewlett Packard and Intel, Microsoft, Wikipedia, Yahoo and Yandex.
Government Institutions and Universities: US National Security Agency (NSA)
, Carnegie Mellon University, TU Dresden, University of Pennsylvania, University of Central Florida, National University of Ireland, University of Missouri, University of Arizona, University of Glasgow, Berkeley University and National Tsing Hua University, University of California, Poznan University, Florida International University, Zhejiang University, Texas A&M University, University of California at Irvine, University of Illinois, Chinese Academy of Sciences, Vrije Universiteit, Engenharia University, State University of New York, Palacky University, University of Texas at Dallas

Do you need help with Hadoop/Mapreduce?

Bookmark to:

This posting gives an example of how to use Mapreduce, Python and Numpy to parallelize a linear machine learning classifier algorithm for Hadoop Streaming. It also discusses various hadoop/mapreduce-specific approaches how to potentially improve or extend the example.

1. Background

Classification is an everyday task, it is about selecting one out of several outcomes based on their features, e.g

• In recycling of garbage you select the bin based on the material, e.g. plastic, metal or organic.
• When purchasing you select the store from based e.g. on its reputation, prior experience, service, inventory and prices

Computational Classification – Supervised Machine Learning – is quite similar, but requires (relatively) well-formed input data combined with classification algorithms.

1.1 Examples of classification problems

• Finance/Insurance
  • Classify investment opportunities as good or not e.g. based on industry/company metrics, portfolio diversity and currency risk.
  • Classify credit card transactions as valid or invalid based e.g. location of transaction and credit card holder, date, amount, purchased item or service, history of transactions and similar transactions
• Biology/Medicine
  • Classification of proteins into structural or functional classes
  • Diagnostic classification, e.g. cancer tumours based on images
• Internet
  • Document Classification and Ranking

  • Malware classification, email/tweet/web spam classification
• Production Systems (e.g. in energy or petrochemical industries)
  • Classify and detect situations (e.g. sweet spots or risk situations) based on realtime and historic data from sensors

1.2 Classification Algorithms

Classification algorithms comes in various types (e.g. linear, nonlinear, discriminative etc), see my prior postings Pragmatic Classification: The Very Basics and Pragmatic Classification of Classifiers.

Key takeaways about classifiers:

1. There is no silver bullet classifier algorithm or feature extraction method.
2. Classification algorithms tend to be computationally hard to train, this encourages using a parallel approach, in this case with Hadoop/Mapreduce.

2. Parallel Classification for Hadoop Streaming

The classifier described belongs to a familiy of classifiers which have in common that they can mathematically be described as Tikhonov Regularization with a Square loss function, this family includes Proximal SVM, Ridge Regression, Shrinkage Regression and Regularized Least-Squares Classification. (note: If you replace the Square Loss function with a Hinge-Loss function you get Support Vector Machine classification). The implemented classifier – proximal SVM – is from the paper Incremental Support Vector Machine Classification, referred to as the paper below.

2.1 training data

The classifier assumes numerical training data, where each class is either -1.0 og +1.0 (negative or positive class), and features are represented as vectors of positive floating point numbers. In the algorithm below are:

D - a matrix of training classes, e.g. [[-1.0, 1.0, 1.0, .. ]]
A - a matrix with feature vectors, e.g. [[2.9, 3.3, 11.1, 2.4], .. ]
e - a vector filled with ones, e.g [1.0, 1.0, .., 1.0]
E = [A -e]
mu = scalar constant # used to tune classifier
D - a diagonal matrix with -1.0 or +1.0 values (depending on the class)

2.2 the classifier algorithm

Training the classifier can be done with right side of the equation (13) from paper

(omega, gamma) = (I/mu + E.T*E).I*(E.T*D*e)

Classification of an incoming feature vector x can then be done by calculating:

x.T*omega - gamma

which returns a number, and the sign of the number corresponds to the class, i.e. positive or negative.

2. Parallelization of the classifier with Hadoop Streaming and Python

Expression (16) in the paper has a nice property, it supports increments (and decrements), in the example there are 2 increments (and 2 decrements), but by induction there can be as many as you want:

(omega, gamma) = (I/mu + E_.T*E_1 + .. + E_i.T*E_i).I*
                 (E_1.T*D_1*e + .. + E_i.T*D_i*e)

where

E.T*E = E_1.T*E_1 + .. + E_i.T*E_i

and

E.T*De = E_1.T*D_1*e + .. + E_i.T*D_i*e

This means that we can parallelize the calculation of E.T*E and E.T*De, by having Hadoop mappers calculate each of the elements of the sums in as in the Python map() code below (sent to reducers as tuples)

2.3 – the mapper

def map(key, value):
   # input key= class for one training example, e.g. "-1.0"
   classes = [float(item) for item in key.split(",")]   # e.g. [-1.0]
   D = numpy.diag(classes)

   # input value = feature vector for one training example, e.g. "3.0, 7.0, 2.0"
   featurematrix = [float(item) for item in value.split(",")]
   A = numpy.matrix(featurematrix)

   # create matrix E and vector e
   e = numpy.matrix(numpy.ones(len(A)).reshape(len(A),1))
   E = numpy.matrix(numpy.append(A,-e,axis=1)) 

   # create a tuple with the values to be used by reducer
   # and encode it with base64 to avoid potential trouble with '\t' and '\n' used
   # as default separators in Hadoop Streaming
   producedvalue = base64.b64encode(pickle.dumps( (E.T*E, E.T*D*e) )    

   # note: a single constant key "producedkey" sends to only one reducer
   # somewhat "atypical" due to low degree of parallism on reducer side
   print "producedkey\t%s" % (producedvalue)

2.4 – the Reducer

def reduce(key, values, mu=0.1):
  sumETE = None
  sumETDe = None

  # key isn't used, so ignoring it with _ (underscore).
  for _, value in values:
    # unpickle values
    ETE, ETDe = pickle.loads(base64.b64decode(value))
    if sumETE == None:
      # create the I/mu with correct dimensions
      sumETE = numpy.matrix(numpy.eye(ETE.shape[1])/mu)
    sumETE += ETE

    if sumETDe == None:
      # create sumETDe with correct dimensions
      sumETDe = ETDe
    else:
      sumETDe += ETDe

    # note: omega = result[:-1] and gamma = result[-1]
    # but printing entire vector as output
    result = sumETE.I*sumETDe
    print "%s\t%s" % (key, str(result.tolist()))

2.5 – Mapper and Reducer Utility Code

Code used to run map() and reduce() methods, inspired by iterator/generator approach from this mapreduce tutorial.

def read_input(file, separator="\t"):
    for line in file:
        yield line.rstrip().split(separator)

def run_mapper(map, separator="\t"):
    data = read_input(sys.stdin,separator)
    for (key,value) in data:
        map(key,value)

def run_reducer(reduce,separator="\t"):
    data = read_input(sys.stdin, separator)
    for key, values in groupby(data, itemgetter(0)):
        reduce(key, values)

3. Finished?

Assume your running time goes through the roof even with the above parallel approach, what to do?

3.1 Mapper Increment Size really makes a difference!

Since there is only 1 reducer in the presented implementation, it is useful to let mappers do most of the job. The size of the (increment) matrices – E.T*E and E.T*D*e given as input to the reducer is independent of number of training data, but dependent on the number of classification features. The workload on the reducer is also dependent on the number of matrices received by the mappes (i.e. increment size), e.g. if you have a 1000 mappers having one billion examples with 100 features each, the reducer would need to do a sum of one trillion 101×101 matrices and one trillion 101×1 vectors if the mapper sent one matrix pair per training example, but if each mapper only sent one pair of E.T*E and E.T*D*e representing all the mappers billion training examples the reducer would only need to summarize 1000 matrix pairs.

3.2 Avoid stressing the reducer

Add more (intermediate) reducers (combiners) that calculates partial sums of matrices. In the case of many small increments (and correspondingly many matrices) it can be useful to add an intermediate step that (in parallel) calculates sums of E.T*E and E.T*D*e before sending the sums to the final reducer, this means that the final reducer gets fewer matrices to summarize before calculating the final answer, see figure below.

3.3 Parallelize (or replace) the matrix inversion in the reduction step

If someone comes along with a training data set with a very high feature-dimension (e.g. recommender systems, bioinformatics or text classification), the matrix inversion in the reducer can become a real bottleneck since such algorithms typically are O(n^3) (and lower bound of Omega(n^2 lg n)), where n is the number of features. A solution to this can be to use or develop hadoop/mapreduce-based parallel matrix inversion, e.g. Apache Hama, or don’t invert the matrix...

3.4 Feature Dimensionality Reduction

Another approach when having training data with high feature-dimension could be to reduce feature-dimensionality, for more info check out Latent Semantic Indexing (and Analysis), Singular Value Decomposition or t-Distributed Stochastic Neighbor Embedding

3.5 Reduce IO between mappers and reducers with compression

Twitter presented using LZO compression (on the Cloudera blog) to speed up Hadoop. Inspired by this one could in the case of high feature dimension, i.e. large E.T*E and E.T*D*e matrices, compress the output in the mapper and decompress in the reducer by replacing base64encoding/decoding and pickling above with:

producedvalue = base64.b64encode(lzo.compress(pickle.dumps( (E.T*E, E.T*D*e) ), level=1)

and

ETE, ETDe = pickle.loads(lzo.decompress(base64.b64decode(value)))

3.6 Do more work with approximately the same computing resources

The D matrix above represents binary classification with a value of +1 or -1 representing each class. It is quite common to have classification problems with more than 2 classes. Supporting multiple classes is usually done by training by several classifiers, either 1-against-all (1 classifier trained per class) or 1-against-1 (1 classifier trained per unique pair of classes), and the run a tournament of them against each other and pick the most confident. In the case of 1-against-all classification the mapper could probably send multiple E.T*D_c*e – with one D_c per class and keep the same E.T*E, the reducer would then need to calculate (I/mu + E.TE).I once and independently multiply with several E.T*D_c*e sums to create a set of (omega,gamma) classifiers. For 1-against-1 classification it becomes somewhat more complicated, because it involves creating several E matrices since in the 1-against-1 case only the rows in E where the 2 classes competing occur are relevant.

4. Code

(Early) Python code of the algorithm presented above can be found at http://code.google.com/p/snabler/ (open source with Apache Licence). Please let me know if you want to contribute to the project, e.g. from mapreduce and hadoop algorithms in academic papers.

5. More resources about machine learning with Hadoop/Mapreduce?

• Apache Mahout – active project that implements (in Java) several machine learning algorithms (also unsupervised machine learning, i.e. clustering)
• Good paper about machine learning algorithms with mapreduce – http://www.cs.stanford.edu/people/ang//papers/nips06-mapreducemulticore.pdf

Bookmark to:

We developed a tool for scalable language processing for our customer Lingit using Amazon’s Elastic Mapreduce.

More details: http://aws.amazon.com/solutions/case-studies/atbrox/

Contact us if you need help with Hadoop/Elastic Mapreduce.

Bookmark to:

Elastic Mapreduce default behavior is to read from and store to S3. When you need to access other AWS services, e.g. SQS queues or database services SimpleDB and RDS (MySQL) the best approach from Python is to use Boto. To get Boto to work with Elastic Mapreduce you need to dynamically load boto on each mapper and reducer, Cloudera’s Jeff Hammerbacher outlined how to do that using Hadoop Distributed Cache and Peter Skomorroch suggested how to load Boto to access Elastic Blockstore (EBS), this posting is based on those ideas and gives a detailed description how to do it.

How to combine Elastic Mapreduce with other AWS Services

This posting shows how to load boto in an Elastic Mapreduce mapper and gives a simple example how to use simpledb from the same mapper. For accessing other AWS services, e.g. SQS from Elastic Mapreduce check out the Boto documentation (it is quite easy when the boto + emr integration is in place).

Other tools used (prerequisites):

• s3cmd – to upload/download files to S3
• Elastic Mapreduce Ruby Client – to launch Elastic Mapreduce jobs

Step 1 – getting and preparing the Boto library

wget http://boto.googlecode.com/files/boto-1.8d.tar.gz
# note: using virtualenv can be useful if you want to
# keep your local Python installation clean
tar -zxvf boto-1.8d.tar.gz ; cd boto-1.8d ; python setup.py install
cd /usr/local/lib/python2.6/dist-packages/boto-1.8d-py2.6.egg
zip -r boto.mod boto

Step 2 – mapper that loads boto.mod and uses it to access SimpleDB

# this was tested by adding code underneath to the mapper
# s3://elasticmapreduce/samples/wordcount/wordSplitter.py

# get boto library
sys.path.append(".")
import zipimport
importer = zipimport.zipimporter('boto.mod')
boto = importer.load_module('boto')

# access simpledb
sdb = boto.connect_sdb("YourAWSKey", "YourSecretAWSKey")
sdb_domain = boto.create_domain("mymapreducedomain") # or get_domain()
# ..
# write words to simpledb
  for word in pattern.findall(line):
      item = sdb_domain.create_item(word)
      item["reversedword"] = word[::-1]
      item.save()
      # ...

Step 3 – json config file – bototest.json – for Elastic Mapreduce Ruby Client

[
  {
	"Name": "Step 1: testing boto with elastic mapreduce",
        "ActionOnFailure": "<action_on_failure>",
        "HadoopJarStep": {
		"Jar": "/home/hadoop/contrib/streaming/hadoop-0.18-streaming.jar",
          	"Args": [
            	"-input", "s3n://elasticmapreduce/samples/wordcount/input",
            	"-output", "s3n://yours3bucket/result",
            	"-mapper", "s3://yours3bucket/botoWordSplitter.py",
            	"-cacheFile", "s3n://yours3bucket/boto.mod#boto.mod",
          	]
        }
  }
]

Step 4 – Copy necessary files to s3

s3cmd put boto.mod s3://yours3bucket
s3cmd put botoWordSplitter.py s3://yours3bucket

Step 5 – And run your Elastic Mapreduce job

 elastic-mapreduce --create \
                   --stream \
                   --json bototest.json \
                   --param "<action_on_failure>=TERMINATE_JOB_FLOW"

Conclusion
This showed how to dynamically load boto and use it to access one other AWS service – SimpleDB – from Elastic Mapreduce. Boto supports most AWS services, so the same integration approach should work also for other AWS services, e.g. SQS (Queuing Service), RDS (MySQL Service) and EC2, check out the Boto API documentation or Controlling the Cloud with Python for details.

Note: a very similar integration approach should work for most Python libraries, also those that use/wrap C/C++ code (e.g. machine learning libraries such as PyML and others), but then it might be needed to do step 1 on Debian AMIs similar to what Elastic Mapreduce is using, check out a previous posting for more info about such AMIs.

Bookmark to:

Back in May 2000 I wrote that “It seems likely that the specialization in the Internet Information Retrieval (IIR) business will continue. Internet information crawling, pre-processing, indexing, searching and presentation requires different types of technologies and know-how, this might create opportunities for new companies specializing in only one step of the IIR “food chain”"

80legs

80legs is a company specializing in the crawling and preprocessing part, where you can upload your seed urls (where to start crawling), configure your crawl job (depth, domain restrictions etc.) and also run existing or custom analysis code (upload java jar-files) on the fetched pages. When you upload seed files 80legs does some filtering before starting to crawl (e.g. if you have seed urls which are not well-formed), and also handles domain throttling and robots.txt (and perhaps other things).

Computational model: Since you can run custom code per page it can be seen as a mapper part of a MapReduce (Hadoop) job (one map() call per page); for reduce-type processing (over several pages) you need to move your data elsewhere (e.g. EC2 in the cloud). Side note: another domain with “reduce-less” mapreduce is quantum computing, check out Michael Nilsen’s Quantum Computing for Everyone.

Testing 80legs

Note: We have only tried with the built-in functionality and no custom code so far.

1) URL extraction

Job description: We used a seed of approximately 1,000 URLs and crawled and analyzed ~2.8 million pages within those domains. The regexp configuration was used (we only provided the URL matching regexp).

Result: Approximately 1 billion URLs were found, and results came in 106 zip-files (each ~14MB packed and ~100MB unpacked) in addition to zip files of the URLs that where crawled.

Note: Based on a few smaller similar jobs it looks like the parallelism of 80legs is somewhat dependent of the number of domains in the crawl and perhaps also on their ordering. In case you have a set of URLs where each domain has more than one URL it can be useful to randomize your seed URL file before uploading and running the crawl job, e.g. by using rl or coreutil’s shuf.

2) Fetching pages

Job description: We built a set of URLs – ~80k URLs that we wanted to fetch as html (using their sample application called 80App Get Raw HTML) for further processing. The URLs were split into 4 jobs of ~20k URLs each.

Result: Each job took roughly one hour (they all ran in parallel so the total time spent was 1 hour). We ended up with 5 zip files per job, each zip file having ~25MB of data (100MB unpacked), i.e. ~4*5*100MB = 2GB raw html when unpacked for all jobs.

Conclusion

80legs is an interesting service that has already proved useful for us, and we will continue to use it in combination with AWS and EC2. Custom code needs to be built (e.g. related to ajax crawling).

(May 2000 – A few thoughts about the future of Internet Information Retrieval)

Bookmark to:

SimpleDB is a service primarily for storing and querying structured data (can e.g. be used for  a product catalog with descriptive features per products, or an academic event service with extracted features such as event dates, locations, organizers and topics). (If one wants “heavier data” in SimpleDB, e.g. video or images, a good approach be to add paths to Hadoop DFS or S3 objects in the attributes instead of storing them directly)

Unstructured Search for SimpleDB

1. Support unstructured search with very low maintenance
2. Combine structured and unstructured search
3. Figure out the feasibility of unstructured search on top of SimpleDB

The Structure of SimpleDB

SimpleDB is roughly a persistent hashtable of hashtables, where each row (a named item in the outer hashtable)  has another hashtable with up to 256 key-value pairs (called attributes). The attributes can be 1024 bytes each, so 256 kilobyte totally in the values per row (note: twice that amount if you store data also as part of the keys + 1024 bytes in the item name). Check out Wikipedia for detailed SimpleDB storage characteristics.

Inverted files

Inverted files is a common way of representing indices for unstructured search. In their basic form they (logically) contain a word with a list of pages or files the word occurs on. When a query comes one looks up in the inverted file and finds pages or files where the words in the query occur. (note: if you are curious about inverted file representation check out the survey - Inverted files for text search engines)

One way of representing inverted files on SimpleDB is to map the inverted file on top of the attributes, i.e. have one SimpleDB domain with one word (term), and let the attributes store the list of URLs containing that word. Since each URL contains many words, it can be useful to have a separate SimpleDB domain containing a mapping from hash of URL to URL and use the hash URL in the inverted file (keeps the inverted file smaller). In the draft code we created 250 key-value attributes where each key was a string from “0″ to “249″ and each corresponding value contained hash of URLs (and positions of term) joined with two different string separators. If too little space per item – e.g. for stop words – one could “wrap” the inverted file entry with adding the same term combined with an incremental postfix (note: if that also gave too little space one could also wrap on simpledb domains).

Preliminary query latency results

Warning: Data sets used were  NLTK’s inaugural collection, so far from the biggest.

Conclusion: the results from 1000 fetches of inverted file entries are relatively stable clustered around 0.020s (20 milliseconds), which are promising enough to pursue further (but still early to decide given only tests on small data sets so far). Balancing with using e.g. memcached could be also be explored, in order to get average fetch time even lower.

Preliminary Python code including timing results (this was run on an Fedora large EC2 node somewhere in a US east coast data center).

Bookmark to:

Sometimes it can be useful to compile Python code for Amazon’s Elastic Mapreduce into C++ and then into a binary. The motivation for that could be to integrate with (existing) C or C++ code, or increase performance for CPU-intensive mapper or reducer methods. Here follows a description how to do that:

1. Start a small EC2 node with AMI similar to the one Elastic Mapreduce is using (Debian Lenny Linux)
• note: We used Alestic’s ami-ff46a796
2. Skim quickly through the Shedskin tutorial
3. Log into the EC2 node and install the Shedskin Python compiler
• Shedskin requires a few libraries: 1) the Boehm-Demers-Weiser garbage collector for C++, 2) PCRE – Perl Compatible Regular Expressions. The Shedskin tutorial for detailed install instructions.
• note: The Alestic Debian AMI is fairly slim, so we had to add some more software make Shedskin work, i.e. GDB
4. Write your Python mapper or reducer program and compile it into C++ with Shedskin
• E.g. the commandpython ss.py mapper.py – would generate C++ files mapper.hpp and mapper.cpp, a Makefile and an annotated Python file mapper.ss.py.
5. Optionally update the C++ code generated by Shedskin to use other C or C++ libraries
• note: with Fortran-to-C you can probably integrate your Python code with existing Fortran code (e.g. numerical/high performance computing libraries). Similar for Cobol (e.g. in financial industry) with OpenCobol (compiling Cobol into C). Please let us know if you try or need help with help that.
6. Add -static as the first CCFLAGS parameter in the generated Makefile to make it a static executable
7. Compile the C++ code into a binary with make and check that you don’t get a dynamic executable with ldd (you want a static executable)
8. Run strip on the binary to make it smaller
9. Upload your (ready) binary to a chosen location in Amazon S3
   • e.g. via commandline with S3CMD, with a UI using S3Fox or Cloudberry S3 Explorer or programmatically with Boto.
10. Read Elastic Mapreduce Documentation on how to use the binary to run Elastic Mapreduce jobs.
    • note: Peter Skomoroch has written a good tutorial for Elastic Mapreduce

Note: if you skip the shedskin-related steps this approach would also work if you are looking for how to use C or C++ mappers or reducers with Elastic Mapreduce.

Note: this approach should probably work also with Cloudera’s distribution for Hadoop.

Bookmark to:

Other recommended writeups :

• Hadoop World NYC (Hilary Mason)
• The View from HadoopWorld (Stephen O’Grady)
• Post Hadoop World Thoughts (Deepak Singh)
• Hadoop World, NYC 2009 (Dan Milstein)
• Hadoop World Impressions (Steve Laniel)

—

Location: Roosevelt Hotel, NYC

1235 Joe Cunningham – Visa – Large scale transaction analysis
– responsible for Visa Technology Strategy and Innovation
been playing with Hadoop for 9 months
probably many in audience learning and starting out with Hadoop

Agenda:
1) VisaNet overview
2) Value-added information products
3) Hadoop@Visa – research results

About Visa:
- 60 Billion market cap
- well-known card products, and also behind the scene information products
- Visa brand has high trust
- For a card-holder a Visa-card means global acceptance
- For a shopowner, if you get a Visa payment aproval you will be payed

VisaNet
VisaNet is the largest, most advanced payment network in the world
characteristics:
28M locations,
130M authorizations/day,
1500 endpoints,
Processes transactions faster than 1s
1.4M ATMs,
Processes in 175 currencies,
Less than 2s unavailability per year (!)
- according to my calculations six 9s (0.999999366)
16300 financial institutions

Visa Processing Architecture
Security/Access Services -> Message|File|Web
VisaNet Services Integration -> Authorization|Clearing&Settlement
Dispute handling, Risk, Information
Scoring every transaction (used for issuer to approve/decline transaction)

Value added Info products
- Info services
Client: Portfolio Analysis, Visa Incentive Network
Accountholder: transaction alerts, accoutnt updater, tailored rewards
- Risk management services
Account monitoring
Authentication
Encyption

Hadoop@Visa
Run a pipeline of prototypes in lab facility in SF
Any technology taken into Visa needs to match scalability and reliability requirements

Research Lab Setup
- VM System:
Custom Analytic Stacks
Encryption Processing
Relational Database
- Hadoop Systems
Management Stack
Hadoop #1 ~40TB / 42 nodes (2 years of raw transaction data)
Hadoop #2 ~300TB / 28 nodes

Risk Product Use Case
Create critical data model elements, such as keys and transaction statistics, which feed our real-time risk-scoring systems
Input: Transactions – Merchant Category, Country/Zip
Output: Key & Statistics – MCCZIP Key – stats related to account, trans. type, approval, fraud, IP address etc.
Research Sample: 500M distinct accounts, 100M transactions per day, 200 bytes per transaction, 2 years – 73B transaction (36TB)
Processing time from 1 month to 13 minutes! (note: ~3000 times faster)
(Generate synthetic transactions used to test the model)

Financial Enterprise Fit
- key questions under research:
- what will the Hadoop Solution Stack(s) look like?
- File system, Transaction Sample System, Relational Back-end (integration path), Analytics Processing
- Internal vs external cloud
- How do I get data into a cloud in a secure way.
- How does HSM and security integration work in Hadoop
- What are the missing pieces?

Why Hadoop@Visa?
- analyze volumes of data with response that are not possible today
- requirement: need to fit with existing solutions

Cross Data Center Log Processing – Stu Hood, Rackspace

(Email and apps division, work on search team)

Agenda
Use Case Backgound
- “Rackapps” – Hybrid Mail Hosting, 40% use a mix of exchange and rackspace mail

Use Case: Log Types

Use Case: Querying
- was the mail delivered?
- spam – why was it (not) marked as spam
- access – who checked/failed to check mail?
more advanced questions:
- which delivery routes have the highest latency?
- which are the spammiest IP?
- Where in the world do customers log in from
Elsewhere:
- billing

Previous Solutions
- 1999-2006 – go to where log files are generated, querying with grep
- 2006-2007 / bulk load to MySQL – worked for a year

Hadoop Solution
- V3 – lucene indexes in Hadoop
- 2007- present
- store 7 days uncompressed
- queries take seconds
- long term queries with mapreduce (6M avail for MR queries)
- all 3 datacenters

Alternatives considered:
- Splunk – good for realtime, but not great for archiving
- Data warehouse package – not realtime, but fantastic for longterm analysis
- Partioned MySQL – half-baked solution
=> Hadoop hit the sweet spot

Hadoop Implementation
SW
- collect data using syslog-ng (considering Scribe)
- storage: deposits into Hadoop (scribe will remove that)
HW
- 2-4 collector machines per datacenters
- hundreds of source machines
20 solr nodes

Implementation: Indexing/Querying
- indexing – uniqe processing code for schema
- querying
- “realtime”
- sharded lucene/solr instances merge-index chunk from Hadoop
- using Solr-API
- raw logs
- using Hadoop Streaming and unix grep
- Mapreduce

Implementation: Timeframe
- development – 1.5 people in 3 months
- deployments – using clouderas distribution
- roadblocks – bumped into job-size limits

Have run close to 1 million jobs on our cluster, and it has not gone down (except for other reasons such as maintenance)

Advantages – storage
- all storage in one place
Raw logs: 3 days, in HDFS
Indexes: 7 days
Archived Indexes: 6 months

Advantages – analysis
- Java Mapreduce API
- Apache Pig
- ideal for one-off queries
- Hadoop Streaming

Pig Example – whitehouse.gov mail spoofing

Advantages – Scalability, Cost, Community
- scalability – easy to add nodes
- cost – only hardware
- community – cloudera has been a benefit, deployment is trivial

Data Processing for Financial Services – Peter Krey and Sin Lee, JP Morgan Chase

Innovation & Shared Services, Firmwide Engineering & Architecture

note: certain constraints what can be shared due to regulations

JPMorgen Chase + Open Source
- QPD (AMQP) – top level apache project
- Tyger – Apache + Tomcat + Spring

Hadoop in the Enterprise – Economics Driven
- attractive: economics
- Many big lessons from Web 2.0 community
- Potential for Large Capex and Opex “Dislocation”
- reduce consumption of enterprise premium resources
- grid computing economics brought to data intensive computing
- stagnant data innovation
- Enabling & potentially disruptive platform
- many historical similarities
- java, linux, tomcat, web/internet
- minis to client/server, client/server to web, solaris to linux, ..
- Key question: what can be built on top of Hadoop?
Back to economics driven – very cost-effective

Hadoop in the Enterprise – Choice Driven
- Overuse of relational database containers
- institutional “Muscle memory” – not too much else to choose from
- increasingly large percentage of static data stored in proprietary transactional DBs
- Over-Normalized Schemas: still Makes sense with cheap compute&storage?

- Enterprise Storage “Prisoners”
- Captive to the economics & technology of “a few” vendors
- Developers need more choice
- Too much proprietary, single-source data infrastructure
- increasing need for minimal/no systems + storage admins

Hadoop in the Enterprise – Other Drivers
- Growing developer interest in “Reduced RDBMS” Data technologies
- open source, distributed, non-relational databases
- growing influence of web 2.0 technologies & thinking of enterprise
- hadoop, cassandra, hbase, hive, couchdb, hadoopDB, .. , others
- memcached for caching

FSI Industry Drivers
- Increased regularity oversight + reporting = More data needed over longer period of time
- triple data amounts from 2007 to 2009
- growing need for less expensive data repository/store
- increased need to support “one off” analysis on large data

Active POC Pipeline
- Growing stream of real projects to gauge hadoop “goodness of fit”
- broad spectrum of use cases
- driven by need to impact/dislocate OPEX+CAPEX
- looking for orders of magnitude
- evaluated on metric based performance, functional and economic measures
- avoid the “data falling on the floor phenomena”
- tools are really really important, keep tools and programming models simple

Hadoop Positiong
- Latency x Storage amount curve,

Cost comparisons
- SAN vs Hadoop HDFS cost comparison (GB/month)
- Hadoop much cheaper

Hadoop Additions and Must Haves:
- Improves SQL Front-End Tool Interoperability
- Improved Security & ACL enforcement – Kerberos Integration
- Grow Developer Programming Model Skill Sets
- Improve Relational Container Integration & Interop for Data Archival
- Management & Monitoring Tools
- Improved Developer & Debugging Tools
- Reduce Latency via integration with open source data caching
- memcached – others
- Invitation to FSI or Enterprise roundtable

Protein Alignment – Paul Brown, Booz Allen

Biological information
- Body – Cells – Chromosomes – Gene – DNA/RNA

Bioinformatics – The Pain
- too much data

So What? Querying a database of sequences for similar sequences
- one-to-many comparison
- 58000 proteins in PDB
- Protein alignment frequently used in the development of medicines
- Looking for a certain sequence across species, helps indicate function
Implementation in Hadoop
- distribute database sequence accross each node
- send query seq. inside Mapreduce (or dist.cache)
- scales well
- existing algorithms port easily

So What? Comparing sequences in bulk
- many-to-many
- DNA hybridiation (reconstruction)
Ran on AWS
Hadoop:
- if whole dataset fit into one computer
- Used distributed cache, assign each node a piece of the list
- But if the does not fit on one computer….
- pre-join all possible pairs with one MapReduce

So What? Analyzing really big sequences
- one big sequence to many small sequences
- scanning dna for structure
- population genetics
- hadoop implementatoin

Demonstration Implementation: Smith-Waterman Alignment
- one of the more computationally intensive matching and aligmnent techniques
- big matrix – (sequences to compare on row and column and calculations within)

Amazon implementation
- 250 machines
- E2
- run in 10 minutes for a single sequence. Runs in 24hrs for NxN comparison
- cost $40/hr

==> very cool 3D video of amazon ec2 nodes
- failing job due to 10% of nodes stuck on something (e.g. very long sequences)

Real-time Business Intelligence, Bradford Stephens

Topics
- Scalability and BI
- Costs and Abilities
- Search as BI

Tools: Zookeeper, Hbase, Katta (dist.search on Hadoop) and Bobo (faceted search for lucene)
- http://sourceforge.net/projects/bobo-browse/
- http://sourceforge.net/projects/katta/develop

100TB structured and unstructed data – Oracle 100M$, Hadoop and Katta 250K$

Building data cubes in real time (with faceted search)

Real-time Mapreduce on HBase
Search/BI as a platform – “google my datawarehouse”

Counting, Clustering and other data tricks, Derek Gottfried, New York Times

back in 2007 – would like to try as many EC2 instances as possible

Problem
- freeing up historical archives of NYTimes.com (1851-1922)
(in the public domain)

Currently:
- 2009 – web analytics
3 big data buckets:
1) registration/demographics
2) articles 1851-today
- a lot of metadata about each article
- unique data, extract people, places, .. to each article => high precision search
3) usage data/web logs
- biggest piece – piles up

How do we merge the 3 datasets?

Using EC2 – 20 machines
Hadoop 0.20.0
12 TB of data
Straight MR in Java
(mostly java + postprocessing in python)

combining weblog data with demographic data, e.g. twitter clicks backs by age group

Bookmark to: