InfoSphere Big Data - A Cursory Overview

After a half day of talking with IBM reps about Big Data products and some use cases, here is how I summarized how the pieces fit together.

At time 0 you collect everything and analyze it for correlations to determine which data items are valuable and how they relate to each other (Big Insights platform). Then you build a control model.

Learning from time 0 is used to configure a "real time" strategy for the data analysis and system control.

• Streams provide real time processing of data "on the wire" - nothing need be stored. The output of this is three fold:
• "Live" reports for users
• A data subset to feed to the data warehouse
• Control signals to feed back to the data collectors to adjust behavior (if needed).
• Netezza (Data Warehouse) provides a location where "fast" analysis on a "limited" subset of the data can occur.

Hadoop holds everything else so that longer term analysis with full data sets is possible. This could be used to:

• Adjust the control models
• Change which data subsets are warehoused 
• Perform ad hoc deep dive analysis. 
• Perform regular analysis on data sets which are too large to reasonably warehouse (e.g. raw scan data).

Yes! We Have NoSQL Today.

Once you have a system which can process huge amounts of data (big data), you need a place to store all of that data. This is what databases are for.

Traditionally, this has meant a relational database. But relational databases place many constraints on how the data is modeled ("normal forms") which are inconsistent with the high volume data sources which need to be analyzed (e.g. all the webpages in the world, all the legal documents in your company or all the tweets being posted each day).

• Relational DBs require that data be modeled into a set of tables that contain unique entities (rows), described by attributes (columns) which are arranged in such a way as to describe one aspect of each entity in each table with no redundancies.
• said another way: 
• each row has a primary key made from one or more columns. Column data contains single values (1NF).
• All columns in a table relate only to the complete primary key (2NF)
• All columns in the table contain data which is not derived from other columns in that table (3NF)
• To add more columns which do not fit these constraints, you must put them in another table and join them together.

                              A good example of what this means is here.

• said yet another way: The key, the whole key and nothing but the key.

These restrictions allow for optimal query structuring and performance while minimizing anomalies due to data changes. However they do not easily support the lack of simple structure between the contents of many data sets.

• Non-Relational (NoSQL) DBs remove the restrictions on data normalization and focus, instead, on optimizing around data that does not fit well into the normalized structure which relational DBs (mostly) require. Because there are different analyses of interest and different data sources which "best" embody the data of interest, there are different types of NoSQL databases.

Below is a diagram showing the various database types.

• Key Value (aka Big Table)
• Data is stored in a GIANT ordered table of rows and columns. 
• Rows and columns still serve the same general purpose as in a relational DB case
• rows = unique entities
• columns = attributes.
• ...but normalization is not required (or expected)...
• Data may be sparsely populated in the columns. 
• I.e. a given row may only have data values for a small fraction of the columns (because most the columns don't apply to the entity this row describes). 
• Columns may be VERY large in number and depend on what the DB is structured to query for.
• e.g. all unique word pairs for the entities in the database
• Google originally developed this technology for searching through web pages to fulfill search criteria. Roughly speaking:
• rows = web pages
• columns = search terms
• Document
• Entire documents are stored in a searchable format.
• JSON (JavaScript Object Notation)
• BSON (Binary JavaScript Object Notation)
• XML (eXtensible Markup Language)
• Queries search through the documents to identify the information of interest and return statistics or the document IDs.
• Good for finding actual documents which contain specific information or summarizing the information contained in a set of documents. 
• Graph
• Stores information about relationships between entities (objects) in the DB
• Good for finding objects that are related to each other according to certain criteria.
• e.g. find people (entities) who are members of the YMCA (another entity) who lived in New York in 1999.

How does this relate to Big Data?
Many NoSQL DBs are built to operate on distributed file systems and process queries via distributed computing. In fact, the very nature of the data being looked at is so large

• operating on >100 PetaBytes and ingesting >500 TeraBytes/day for Facebook
• processing >20 Petabytes / day for Google
• Processing > 340 Million tweets / day for Twitter (~44GB/day but a huge number of entities)

 and so unstructured (hard to normalize) that the data could be stored and handled in no other way.

A little bit about Big Data - Hadoop

One of the ways that Big Data of the sort discussed in this last post is implemented is with an open source technology stack called Hadoop.

Hadoop consists of a two main parts:

• HDFS - Hadoop File System
• MapReduce infrastructure

These allow data processing jobs to be divided among multiple nodes and then aggregated into a single result. In essence, this constructs a large, parallel computer from many smaller computers - basically the opposite of virtualization.

Schematically, a Hadoop cluster looks like this:

10,000 ft view of how it works

• The Job Tracker on the master server gets a job
• The Job Tracker breaks up the job using the map function
• Basic queuing ensures that any one node is not overloaded with tasks
• The are tasks preferentially distributed to the nodes nearest the data on which the task must operate to minimize file transfer overhead*.
• nearest = same node as the data resides.
• next nearest = different node but behind the same switch (so that data transfer is localized to that network segment).
• The Job Tracker gets status for all tasks via the Task Trackers as they run. 
• If a node stops reporting, the Job Tracker will redistribute that node's tasks to another node.
• When all tasks are complete for a job, the Job Tracker has the nodes execute the reduce function to generate a single result from the tasks' output.
• The final output may then be used by other applications directly or as the input to another MapReduce iteration.

I am certain that I missed almost every important detail in terms of the actual engineering implementation of a Hadoop cluster. But for those who just need a cursory understanding of the technology in order to make sufficient sense of what engineering is actually talking about to sanity check proposals,  I hope this hits the mark.

---------------------------------------------------------------------------------------------------
* HDFS and the awareness of data's physical location is very important in dealing with large data.
To make compute and storage capacity scale linearly with cost, the data must be spread around all the nodes in the cluster and a record kept of where all the data actually is. It can't be kept centrally or data transfer becomes a serious bottleneck to computational performance. I.e. it could take longer to move the data to a compute node over a network than it takes to actually process it. 
This distribution is managed by HDFS.

HDFS holds and replicates data in the system in order to minimize the chance of a bottleneck. It does this by keeping at least 3 copies of the data:

• an original
• a copy on another node behind the same switch as the original
• a copy on another node on a different switch from the original

This replication attempts to strike a balance between:

• keeping enough copies of the data to minimize the queue size on each node (to ensure timely completion of jobs) and to ensure robust execution despite failed tasks on dead nodes 

-and-

• reducing the total storage capacity of the system by duplicating data.