Sunday, October 13, 2013

Yesterday I woke up and realized I was a Product Owner.

Well, not yesterday. But I have had a recent revelation that I am no longer an engineer. And I'm not sure if I can ever really go back, even if I were to write production code again (I'm always writing code, but at this point it's more of a hobby than anything).

My journey into product ownership began innocently enough when I assumed technical leadership for a set of services built around Hadoop and other NoSQL Platforms -- the Enterprise Data Platform I've posted about before --  and has taken over a year. In that year I have transitioned from a person that provides purely technical solutions to a product owner. What's the difference? To me it is simple: a technologist implements solutions the best way they possibly can to address a perceived problem. A product owner makes sure that the  problem being solved actually matters to someone. Preferably before the technologist gets too far down the solution path.

In my last post I talked a lot about secondary vs primary value propositions. The Enterprise Data Platform I have been working on as a technologist is a group of services have secondary value propositions. The technologies used to solve them are definitely awesome.  But that's irrelevant. As a product owner I've got to make sure that those problems are worth solving to someone, because there is a significant investment of time and resources required to solve problems. Even if the solutions to those problems don't end up getting used. 

Nothing is more frustrating than burning effort on a solution that goes nowhere. Especially if that time was spent crafting a robust, scalable, well tested, well documented implementation of that solution. It's frustrating because that work resulted in a solution that either no one understood or no one wanted. In fact it was my frustration with having been through the good solution/bad product cycle several times that made me take the jump from technologist -- one of the people that implements solutions -- to product owner -- one of the people who decides whether implementation is worth it.

One of the hardest things about transitioning to a product owner from a purely technical role is being able to distinguish whether I should be doing something because I can, or because it solves a problem that people actually have. This is hard for me because as a technologist I tend to get sucked into solving hard problems before asking whether those problems should actually been solved. Ironically I've found that the best technologists I've worked with are the ones that stand back, ask the bigger picture questions, and use the answers to implement elegant, concise solutions.

Product owners need to do the same thing. I think the main job of a product owner is to learn what product the customer base really wants. Instead of plowing ahead with user stories and a feature roadmap, the best product owners I've seen are the ones that can step back from that process and ask the bigger picture questions, and use the answers to make a better product.

Answering big picture questions correctly allows product owners to fail fast -- throw away features that aren't working and iterate towards ones that do. And before doing any of that, the best product owners I know are the ones that can articulate the real value proposition of any effort before embarking on it. They may not know what or how they are going to deliver the effort, but they've nailed why the effort is worth making or not.

In my Product Owner self education, I've learned (from lots of reading and even more bleeding)  that clearly articulating the value proposition and true effort cost is critical to understand whether the effort is worth exploring and sell that exploration process to the people financing it and the people working on it. One of the best things I've found to help clarify product vision is this product canvas from Shardul Metha.


I've been going through our product suite with this product canvas. That process, while painful, has clarified why we should continue doing certain things and why we should stop doing others.

The interesting thing about the canvas is that everything is linked.  Here are some examples of what  I mean:

  1. If Key Success Metrics don't link back to the Value Proposition through the Solution, it doesn't matter how obvious they are, they're wrong. 
  2. If you don't have viable Channels, no executive is going to sign up to be a Key Stakeholder, and it's going to be hard to find a customer that could be a champion. 
  3. If the Value Proposition doesn't address the problems that customers have more efficiently than the alternatives, the Business Value of the effort is weak. 
  4. An ill defined or prohibitively expensive Cost Structure can weaken the best Value Proposition as well because Business Value will be reduced.


Because everything is linked, I find myself revisiting what I had thought was 'obvious' or at least 'set'. Like the Value Prop. Or the Customer. Or even the Problem. But after several cycles what comes out of the other side is either very strong or very weak. This makes deciding which efforts to pursue much easier.

Back to my original problem: I'm selling secondary solutions. The product canvas helps by allowing me to clarify the customer I'm helping, the problem they have, my specific value proposition, and how I would go about fulfilling that value proposition. I've found that for these secondary solutions, linking them to primary solutions is critical. An example:

We've been working on a reliable data delivery service that allows engineering teams to stream their data into different endpoints. Teams can build in 'routing directions' to the data they send that will enable it to land in HDFS, Mongo, Cassandra, ElasticSearch, and other endpoints. That is technically very cool. But we have not been able to clearly articulate why we would do something like this to our leadership because as technologists, the advantage of having something like this as a service is obvious. Namely, people could reuse this service instead of building a custom one.

As a product owner I get a chance to look at the solution from another point of view. What problem am I trying to solve?  How much does it cost to stand up and run? Are there commercial alternatives that we would be better of using? Can I find someone that is willing to try it and work with them to load test it? What other teams do I need to deliver it? Who do I need to sponsor it? And how am I going to get people to use it? For the a secondary solution like this, perhaps the most important question to answer is: what primary problems would be easier to solve if this existed today, and how does that benefit the business? 

None of those questions have anything to do how we are solving the problem. But they are the minimum set of questions that need to be answered to justify continuing this effort. And, had I been as educated when we had started the effort, this is the minimum set of questions that we would have asked prior to doing any work. The anwers to these questions laid out in product canvas form would have been a decision making compass -- when confronted with choice A vs choice B, the product canvas would have given us the tools to make the best decision.

I think that this approach is fundamentally right, but it is one that I am only just starting. The product canvas approach -- clarifying why before what and how--  is obviously the first foundational step in delivering a valuable product. Our teams are going through this journey, and my hope is to write down what is working, and what isn't, as we now try to deliver products whose value proposition we clearly understand :) 

Thursday, September 26, 2013

Enterprise Data Platform: A Reboot and a Reality Check

The last post I wrote on the Enterprise Data Platform was in January. It's September. What happened?

A whole lot, actually. My understanding of what an Enterprise Data Platform is and how it needs to be 'sold' has changed dramatically. My role in this process has also changed. And my understanding of delivering software and running a team has grown and changed  and continues to change for the better.

What I've found out in the past year, among many other things, is that getting people to fund what I've been calling an Enterprise Data Platform is as much about education as it is about technical execution. I'm not talking about educating other people, I'm talking about educating myself. It has been an incredibly educational, humbling, uncomfortable, frustrating, awesome nine months.

When I look back at the posts I wrote early this year, one thing that is very murky throughout a reasonably well laid out argument for data management is a value proposition. I didn't know that when I wrote it, but I quickly found it out when I went to ask for money.

Here is the value disconnect: a system to collect, manage, and leverage data is that system solves  a secondary problem that assumes that a primary problem has been solved. In other words: If I'm trying to get you to fund a data collection, storage, and management platform, I'm assuming that there is something that is already generating the data that needs to be stored.  Netflix, Amazon, Google, my bank website, Blogspot, all solve primary problems. Those problems are easy to explain, regardless of how hard they are to implement. Solutions to primary problems have clear, concise, direct connections to value.

Solutions to secondary problems are optimizations of primary solutions. Decreasing time to insight on operational metrics of a website is an optimization. A great one, to be sure, but not necessary if the website is not getting any traffic.

Any solution to a secondary problem has an indirect value connection at best. Secondary solutions only make sense when the initial value proposition of the primary problem is diluted or reduced due to a secondary problem. A system that doesn't scale to support a site whose popularity is exploding through the roof is a secondary problem. The Secondary problems I see in my current role are operational in nature, and the solutions to them are optimizations. They can deliver huge value when done correctly.

"When done correctly". Three words that are seared into my brain. In the past year several things have happened while I've been trying to explain, again and again, why we should build a solution to a secondary problem, and while I've been trying to build that solution with limited resources:
  1. I've realized that the best way to solve a secondary problem is one primary problem at a time. Building a platform to optimize an undefined set of primary solutions is a risky, 'field of dreams' approach, and there are many ways to go awry. 
  2. I've become less of a technologist and more of a product owner. My last piece of production facing code will (hopefully) be retired in the next couple of months. There are much better engineers on the team, and I rely on them to deliver working software in the same way that they rely on me to come up with a useful product.
  3. Where I used to think about use cases and requirements and assume that these were valid, I now question and validate product direction up front. That involves use cases, but if the use case is invalid, why spend time extracting requirements from it? I spend more time thinking about validating the use case as cheaply as possible.  Requirements emerge and solidify as product direction takes shape -- doing them in advance of having a validated use case seems backwards. This insight radically changes the way software is delivered, and our teams are in the middle of this change process.
  4. The cost and recovery plan for any effort -- infrastructure and resources -- combined with time to recovery, is best defined as soon as you have validated use cases. Those plans need to change as validated features emerge that impact cost and recovery. In previous roles I had been 'sheltered' from that aspect of the business. I'm finding now that financial data is the ultimate data point that helps quantify whether value is being delivered.
This process is far from complete. I am continuing to learn every day, and while it can get very uncomfortable, it has been an amazing education. 

I've tried to write things down several times in the past 9 months. I haven't gotten very far because what I was writing didn't feel complete. Writing about the technology is only one side of the story. What I've learned in the past nine months is that there is a much bigger picture -- now that I'm starting to be able to externalize what I've learned, I'm excited to write about it. More soon...

Friday, January 11, 2013

More on the Enterprise Data Platform: Data Requirements

In my last post, I talked in very general terms about an Enterprise Data Platform (EDP) and in very specific terms around what I consider to be a core requirement of any EDP, data governance. If I have a set of services and processes that provide data governance, I have a way to manage data. What kind of data am I trying to manage?

I'm primarily concerned with building systems that contain  event data and reference data. Event data can be data copied from OLTP systems, it can be user click streams, machine data collected at regular intervals --  anything that signals an event happening.  Event data can be huge.

Reference data is data that can be used to classify/quantify aspects of an event.  If I'm looking at a click stream, a user ID is reference data that I can aggregate events by. In OLAP terms, events can be cast as facts, with reference data providing some of the dimension values.

This data starts to get valuable when  'raw' event data is joined to reference data, and then in turn joined to other event data along a specific reference dimension. For example, aggregating user click stream and email campaign opens by user ID could be used to track the rate at which the email campaign actually generated new users.

In addition to that kind of analytical use, event data can be used to classify users and/or determine their affinities. This kind of derivative data is typically referenced at runtime, by the same applications that are generating the event data.

The two cases above are interesting because they have opposing requirements. Analytic data must first and foremost be consistent, especially when financial reporting and/or business decisions are being made from that data. Consistency in this case implies that when a value is written, the next read reflects the last change made to that variable.

Runtime data must first and foremost be available, because unavailability of data may compromise application behavior. Besides, preference or personalization data is derivative data, generated on a set interval as a batch process. Being out of sync for a long time will eventually mean the application will grow 'stale', but when we talk about availability at the expense of consistency, the usual case is that any inconsistency between read state and written state will be resolved in sub second time.

Why is this important?

  1. Enterprises collect a lot of this data -- TB/day -- and that scale will swamp any single machine based database. 
  2. The enterprise must therefore store data on a storage platform that spans many machines -- a cluster. 
  3. The moment data is stored in a cluster, it is subject to the CAP theorem, which states that a distributed system cannot enforce Consistency, Availability, and Partition Tolerance at the same time. 
  4. While it is valid to desire a clustered system that favors either Consistency or Availability, Enterprise level  requirements state always require partition tolerance, so we can have one or the other, but not both. 
  5. This means that there are usually two main systems, an analytic focused system, and a runtime facing system. The analytic focused system favors consistency, the runtime facing system favors availability.
  6. An EDP that wants to offer both analytic and runtime facing data must have at least two systems, one which is consistent for analytic data, the other which is available for runtime facing data.
So now my definition of an EDP has evolved. It must have some fundamental Data Governance, and due to the scale enterprises operate at, must have 1..N analytics focused platform(s) and 1..N runtime facing platform(s). In my next post I'll focus on the analytic focused platform requirements. 







Friday, November 16, 2012

What is an Enterprise Data Platform, Anyway?

I've been trying to write about my requirements for a Data Platform for the last month. The problem is that I was trying to write this at the exact same time that my understanding of both the requirements and the Platform was shifting, so  I ended up writing the requirements, coming back to them a week later, throwing them all out, and rewriting them again. This is revision number three, and I actually feel that I've stabilized my definition. The biggest change between now and a month ago is that I realized that I am actually thinking about an Enterprise Data Platform, which is different than a Data Platform.

First of all: What is a Data Platform? Here is a first crack at a definition: a Data Platform enables people to leverage data by facilitating data collection, storage, transformation, and access. In other words, I need to be able to get data in, persist it, access it, change it, and still access it. All of that is completely possible with {name your nosql storage engine of choice}. So why am I not calling {name your nosql storage engine of choice} an Enterprise Data Platform?

I have a short answer that is explainable with a longer answer. The short answer: Enterprise Data Platforms have another requirement: the ability to manage all of that data. 

The long answer: my experiences with databases and persistence technology started from the perspective of a developer in a startup, not an administrator or data modeler in an enterprise. I wanted to successfully collect, store, transform, and retrieve data, in order to fulfill the functional requirements of the applications I wrote. I found that a lot of the database admins and data modelers/architects I worked with actually got in the way of this fairly simple mission.

A lot of the constructs and restrictions imposed on the data model and the database by both parties seemed arbitrary, and when I asked them for their reasoning, their answers had no relevance to my single data set use case.  So, when I discovered Hadoop while working around a complex set of ETLs that had ground our database to a halt in 2009, I was elated, because both parties were sidelined and I could do what I needed to do. Their apoplectic reactions to 'schema on read' used to fill my heart with joy.  That makes my current perspective very ironic.

The change in perspective happened when I changed jobs. I went from being a developer that used Hadoop to deliver solutions to an architect/product manager position where I led a team that operationalized Hadoop, Cassandra, Mongo, and Solr and then started to store and curate enterprise level data as a centralized service offering. The moment I started doing thinking past pure functionality and more about operations, I started to care about a lot of the things that people who manage data -- the people I used to disparage -- have been caring about for a while.  The moment I became a data curator or steward of a set of very diverse and valuable data, I started to see real value in Data Governance, which was a term that previously made me roll my eyes in impatient disgust.

Data Governance is defined in wikipedia as a set of processes around "data quality, data management, data policies, business process management, and risk management surrounding the handling of data in an organization".  As a developer in a startup tasked with doing a few things very well, I could care less about data governance. I know my data, it may be vast but it has the same schema, which I know and can code for.

However, as a member of a large enterprise storing many kinds of data who wants to use some of that data, I now must rely on data quality, definition, and security. Without those three I may be able to generate some value from that data, but the value is undermined by the lack of defined quality of the inputs, the lack of defined standards to normalize the data to, and the lack of security which means that the data could be compromised by a rogue user/process. Those 'constraints' on enterprise data are in place because they have a direct impact on the bottom line.

Breaking down Data Governance from the above definition, I get the following:
  1. Data resources must be discoverable: there must be a set of defined metadata that analysts can search for data by. 
    • common metadata is most effective when there is a common data model, which becomes hard to enforce when an enterprise spans many different business units. 
  2. Data structure must be describable so that analysts can consume it. 
  3. Data Quality must be known and identified every time new data is ingested. 
    • When I load data from an external source, downstream processes rely on it conforming to a schema. I need to score the data by how much of it conforms to that schema.
    • When Data Quality dips below a defined level, it must be treated as an operational issue. 
  4. Data Replication must be defined and enforceable.
    • Storage systems must allow users to set a replication factor to account for storage failures. 
  5. Data Replication must be defined, enforceable, and applicable per unique data type.
    • Replicating full data sets may be a business requirement: if so, a Data Platform must replicate data along the following dimensions: resource location (where) and range (how much). 
  6. The context of the data, e.g. what the units of the data are, must be defined in order for an analyst to successfully trust and consume the data. 
    • Unitless data is much less useful. Note that this implies common units across an enterprise, which is harder to do than one would think. Defining and encouraging a Master Data Model helps restrict units and meaning to allow different people to utilize data that they haven't produced. 
    • Data that has been vetted is 'trusted' which means that someone is standing behind it. Standards around trust are important when you have many owners of data. They imply standards around data quality and context. 
  7. Data must be secure
    • Access must be restricted to specific owners. 
    • Those owners must be able to centrally manage permission granting to share data with other users. 
    • Users must be authenticated to the overall system, and can only access data that the data owners have authorized them to. 
    • All actions must be audited.
These requirements are foundational to a data platform that houses many diverse data sets. You could build one without them, but it would only work for a limited set of data. Which is why startups don't care about data governance, and why most NoSQL products are only now starting to think about security. They don't need to -- and to be clear, that's perfectly fine. But that doesn't work or scale at the enterprise level.

What would an Enterprise Data Platform that implemented the requirements above enable? It would enable qualified, standardized, and secure use of data for both analytics and runtime consumption. As a company ran different kinds of analytic efforts on it's data, and generated runtime models from those analytic efforts, other analysts could reuse the input, intermediate, and generated data in different, unanticipated, "recombinant" ways because they can trust the data and can apply company standards to it. The platform would become an enabler, allowing analysts to discover new insights by removing the overhead of managing the data.

I have other requirements of a Data Management Platform besides management, but management is a foundational aspect when managing very many, very diverse sets of data across multiple storage platforms. I hope to start addressing those shortly.









Thursday, July 12, 2012

Calculating Conditional Probability with MapReduce part 4: Applying Correlation and Independence to get better results

In my last post I discussed using mapreduce to calculate item to item conditional probabilities, in other words, the probability that you will click on page B given that you have clicked on page A.

In this post I'm going to discuss how to determining whether a user click on page A is independent of a user click on page B. I went over this in the first post, but it's worth revisiting because taking the theory literally leads to confusion about why it is relevant for recommending products or content.

Independence and Correlation

Some review of the theory using common sense: if I roll two dice together, it's obvious that the probability of getting a 6 on the first dice is independent of the probability of getting a 6 on the second dice. If, on the other hand, I have 4 blue balls and 4 red balls in a drawer, and retrieved a blue ball the first time I reached for a ball, the probability of pulling out a red ball has gone up because now there are 3 blue balls and 4 red balls left. In the first scenario,

P(roll a 6 with dice A) | P(roll a 6 with dice B) = P(roll a 6 with dice A)

in the second: 

P(red ball) | P(blue ball) = # of red balls/(# of red balls + # of blue balls left). 

Given that I've reduced the number of blue balls, P(red | blue) is > P(red in the original case), I'm more likely to pick a red ball if I've picked a blue ball first.

How does that translate into whether to recommend page B given the user has viewed page A? URIs  aren't balls in a drawer. The theory of independent events vs dependent events doesn't really apply to page views...until you consider that the implications of using the independence test actually make sense when deciding whether to recommend page B given the user is viewing page A.

Given that we know P(B | A) and P(B), and we know that 
  1. for independent events P(B | A) = P(B)
  2. for negatively correlated events P(B | A) < P(B)
then any calculated  P(B | A) significantly <= P(B) means that recommending page B given the user is on page A is likely to disengage the user, because many more viewers are likely to view page B without viewing page A than they are to view page B after viewing  page A. 

Calculating Probabilities Again
From above, we need to calculate P(B | A) and P(B) to determine if B should be recommended when the user is viewing A. We have P(B | A) from the last post: the output of our last reducer was 

    uriX     uriY=P(Y | X) uriZ=P(Z | X)...


As for calculating raw probabilities, we can do this by taking the output from our group-by-user mapreduce from the second post, which looked like

   user1  pageA:num_views_A_by_user1  pageB:num_views_B_by_user1...

A map job that submitted key:value pairs from each line of that output, and a reduce job that grouped by (page uri) key and summed the counts would give a total count of page views per page.

public static class AggregateViewCountsMapper extends
   Mapper {

    @Override
    public void map(final LongWritable key, final Text value,
    final Context context) {
 // assume output of GroupViewsByUser:
 // user1\turi1:count1\turi2:count2...\turiN:countN
   
 String raw = value.toString();

 String fields[] = raw.split("\t");
   
 long totalCount = 0;
 for(int i = 1; i < fields.length; i++ ) {
   
         String rawFieldData = fields[i];
         String tokens[] = rawFieldData.split(":");
    
  if(tokens.length != 2) {
  context.getCounter(AggregateViewCounters.BAD_URI_SUBFORMAT).increment(1);
  continue;
  }
  // we care about both tokens
    
  try {
          String count = tokens[1].replaceAll("\\s","");
   Long actualCount = new Long(count);
   context.write(new Text(tokens[0]), new LongWritable(actualCount));
   totalCount += actualCount;
  } catch (IOException e) {
   context.getCounter(
    AggregateViewCounters.MAP_CONTEXT_WRITE_IO_EXCEPTION)
     .increment(1);
    e.printStackTrace();
   } catch (InterruptedException e) {
    context.getCounter(
    AggregateViewCounters.MAP_CONTEXT_WRITE_INTERRUPTED_EXCEPTION)
       .increment(1);
    e.printStackTrace();
    }
   }
   
   // write total counts to a guaranteed unique key
   try {
    context.write(new Text(UNIQUE_KEY), 
                                      new LongWritable(totalCount));
   } catch (IOException e) {
    context.getCounter(AggregateViewCounters.
                                     MAP_CONTEXT_WRITE_IO_EXCEPTION)
         .increment(1);
    e.printStackTrace();
   } catch (InterruptedException e) {
    context.getCounter(AggregateViewCounters.
                                     MAP_CONTEXT_WRITE_INTERRUPTED_EXCEPTION)
         .increment(1);
    e.printStackTrace();
   }
   

  }
 }
    }
}


A special key (UNIQUE_KEY above) could be used to sum up all views per user.

The reducer is a simple aggregator, the final output would look like this:

pageA  total_view_A_ct
pageB  total_view_B_ct
....
UNIQUE_KEY total_view_all_pages_ct

The number of lines of output is equal to the total number of pages in your candidate set+1.

Notes on Final Calculations

I'm leaving this one as a mental exercise:
If the data set is small (<10k pages), you could write some script to calculate probabilities of pages A..N by dividing 

total views of pageA / pageTotal count
..
total views of pageX / pageTotal count

with those counts in place, you can now build a model of views and keep each item B if 

   P(B | A) > P(B)

Note that building the model consists of getting the item:item output, and filtering items as above.

If the data set is larger, and the overall size of the set spans several blocks, then processing the results via mapreduce becomes cumbersome if you don't know the total view count of all pages. I think that this last step definitely requires storing the raw probabilities in a lookup table, then processing the conditional probabilities and filtering out the lists by doing the comparision between P(B | A) and P(B).

How to build the model and store the data above for a more optimal retrieval during model build time is definitely a completely new blog post :)  

Sunday, June 10, 2012

Calculating Conditional Probability via Mapreduce part 3: item to item mappings

In my last post I gave an example of code that grouped views by user ID. In this post I'm going to show how to take the output of that mapreduce and generate item to item mappings that show the conditional probability of URIs being clicked on, given that a specific URI was clicked on.

The output from the views-by-user-ID reducer had the key, followed by a colon, followed by tab delimited URIs and counts (colon separated).

user1\turiX:3\turiY:4\turiZ:12....
user2\turiW:5\turiQ:5
...

We want to generate output that shows the following: given that the user has clicked on X, how many times did they click on Y vs Z, or P? In this output, we are not concerned with specific users, but user behavior in general. We want to sum up general conditional probabilities.

In the mapper, our input is as above: a more concrete example is :

user1\thttp://foo.com/page1:12\thttp://foo.com/page2:23\thttp://foo.com/page3:3

In the mapper we generate permutations of the different key pairs (and ignore the counts for now). In the example above we would generate:

key=>value
http://foo.com/page1=>http://foo.com/page2
http://foo.com/page1=>http://foo.com/page3
http://foo.com/page2=>http://foo.com/page1
http://foo.com/page2=>http://foo.com/page3
http://foo.com/page3=>http://foo.com/page1
http://foo.com/page3=>http://foo.com/page2

Here is the mapper code to generate the permutations
public static class ItemToItemGroupingMapper extends
   Mapper {

  @Override
  public void map(final LongWritable key, final Text value,
	final Context context) {
        // this job takes the input from GroupViewsByUser
	// user1\turiX:3\turiY:4\turiZ:12
	String raw = value.toString();
			
	String fields[] = raw.split("\t");

	if(fields.length == 1) {
		context.getCounter(ItemToItemCounters.BAD_MAP_INPUT_FORMAT)
		.increment(1);
		return;
	}
	// fields[0] is the user ID, we skip that for item-item co-occurrence.
	for(int i = 1; i < fields.length;i++) {
		// permute all combinations of co-occurrences
				
		for(int j = 1;j < fields.length;j++) {
			if(j == i) {
				continue; // dont count self co-occurrences
			}
			else {
		 	    String partsI[] = fields[i].split(GroupViewsByUser.INTERNAL_DELIMITER);
			    String partsJ[] = fields[j].split(GroupViewsByUser.INTERNAL_DELIMITER);
						
			    if(partsI.length != 2 || partsJ.length != 2) {
				context.getCounter(ItemToItemCounters.BAD_MAP_INPUT_FORMAT)
				.increment(1);
				return;
			}
		        try {
			    LOGGER.info(partsI[0]+"=>"+partsJ[0]);
			    context.write(new Text(partsI[0]),new Text(partsJ[0]));
			} catch (IOException e) {
			    context.getCounter(ItemToItemCounters.MAP_CONTEXT_WRITE_IO_EXCEPTION)
				.increment(1);
			    e.printStackTrace();
			} catch (InterruptedException e) {
			    context.getCounter(ItemToItemCounters.MAP_CONTEXT_WRITE_INTERRUPTED_EXCEPTION)
				.increment(1);
			    e.printStackTrace();
			}
		}
					
					
	}
  }
}

In the reducer, these pairs would show up and we can aggregate the counts of each URI. Again, we're not factoring in a normalized weighting at map time. In other words, from the example above, we're not factoring in the 12 hits of page1 vs the 23 hits of page2 when doing co-occurrences, because, as noted above, we're ignoring counts and focusing on co-occurrences

We could do some calculations on these pairs at reduce time -- for a specfic URI X, we would sum total co-occurrences per unique URI Y and divide by total co-occurrences of all URIs to get a number that represented the probability of the total co-occurrences that X and Y made up. We could then reverse sort the URIs by co-occurrence probability to produce output like this for uriX:

uriX\turiY=P(Y | X)\turiZ=P(Z | X)


a real version would look like:

/foo/bar.html   /foo/baz.html=0.5       /foo/car.html=0.5

As expected, in the example above, the probabilities all sum to 1.0.

The reducer code to aggregate counts is below. Its more complex than the previous GroupViewsByUser code because of the probability calculation and reverse sorting.


public void reduce(Text key, Iterable input, Context context) {

        Map URItoCount = new HashMap();
	Iterator it = input.iterator();
	long totalCount = 0;
	// these are co-occurrences: sum them.
	while (it.hasNext()) {
		Text value = new Text(it.next());

		Integer count = URItoCount.get(value);

		if (count == null) {
			URItoCount.put(value, Integer.valueOf(1));
		} else {
			URItoCount.put(value, Integer.valueOf(count + 1));
		}

		totalCount++;
	}

	// emit tab delimited and from greatest to least
	// uri2=prob(uri2 | uri1) uri3=prob(uri3 | uri1)

	Map> byProb = new HashMap>();
	List sortedSet = new ArrayList();
			
	for (Map.Entry entry : URItoCount.entrySet()) {
		double probability = ((double)entry.getValue())/totalCount;
		List list = byProb.get(probability);
				
		if(list == null) {
			list = new ArrayList();
			byProb.put(probability,list);
			sortedSet.add(probability);
		}
		list.add(entry.getKey());
	}

	// sort from greatest to least.
	Collections.sort(sortedSet);  
	Collections.reverse(sortedSet);
			
	// emit uri:value where value = count / total 
	StringBuffer sbuf = new StringBuffer();
			
	for(double probability : sortedSet){
		List uris = byProb.get(probability);
				
		for(Text uri : uris) {				      
                    sbuf.append(uri.toString()).append(INTERNAL_DELIMITER).append(probability).append(FIELD_DELIMITER);
                }
	}
				
	// convert list to hadoop Text
	String coOccurrenceBuffer = sbuf.toString();
	LOGGER.debug("reduce: key = " + key.toString() + ", value = "
			+ coOccurrenceBuffer);

	Text allViews = new Text(coOccurrenceBuffer);

	try {
		context.write(key, allViews);
	} catch (IOException e) {
		LOGGER.error(e);
		context.getCounter(
				GroupByUserCounters.REDUCE_CONTEXT_WRITE_IO_EXCEPTION)
				.increment(1);
	} catch (InterruptedException e) {
		LOGGER.error(e);
		context.getCounter(
				GroupByUserCounters.REDUCE_CONTEXT_WRITE_INTERRUPTED_EXCEPTION)
				.increment(1);

	}
}


Are we done yet? Well....no. While it's nice to see conditional probabilities, they dont tell us anything about whether any two URIs are correlated. That's next. Sample code can be found at https://github.com/arunxarun/cpsamples.

Tuesday, May 15, 2012

Calculating Conditional Probability via Mapreduce part 2: group by userID

In my last post I walked through the basics of conditional probability in order to get around to actually discussing how I would implement it using mapreduce.

The whole calculation of Conditional Probability is a little contrived -- I admitted that up front -- my goal was to provide a 'non word count' example of mapreduce.  It also requires multiple mapreduce steps. I'll  provide the context around those steps to show how they link together to  calculate conditional probability and correlation, both of which are required if we are going to recommend or avoid recommending products when a specific product is shown.

The entire source for all samples (as I write them) can be found at:

https://github.com/arunxarun/cpsamples


Starting Assumptions: the User View Log.
Log format is dependent on who wrote the application, but if you were using something like Common Log Format, you would see the following kind of log:

127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] "GET /products/p1234.html HTTP/1.0" 200 2326

where:

  1. frank is the userid
  2. The timestamp is the bracketed next field
  3. The quoted string afterword contains the HTTP verb plus the uri plus the protocol version
  4. The next field contains the HTTP status code
  5. The final field denotes the amount of time in milliseconds that the request took.

Mapreduce 1: group by UserID
The first mapreduce will group all pages by user during the map phase and sum page counts by user during the reduce phase. It does this by parsing the input lines and emitting the user and URI as a pair into the context:
    
/**
 * the key the user, the value is the URI
 *
 */
public static class GroupViewsByUserMapper extends Mapper {
 
  @Override
  public void map(final LongWritable key, final Text value,  final Context context) {
    // assume common log format: 
    //127.0.0.1\t-\tfrank\t[10/Oct/2000:13:55:36 -0700]\t"GET /apache_pb.gif HTTP/1.0"\t200\t2326\n
    String raw = value.toString();
  
    String fields[] = raw.split("\t");
  
    if(fields.length < 6 || fields.length > 7) {
        context.getCounter(GroupByUserCounters.BAD_LOG_FORMAT).increment(1);
        return;
    }
  
    String userIdText = fields[2];
    String rawHttpInfo = fields[4];
   
    String uriComponents[] = rawHttpInfo.split(" ");
  
    if(uriComponents.length < 3) {
         context.getCounter(GroupByUserCounters.BAD_URI_SUBFORMAT).increment(1);
         return;
    }
    Text userId = new Text(userIdText);
    Text uri = new Text(uriComponents[1]);

    try {
       context.write(userId, uri);
    } catch (IOException e) {
         LOGGER.error(e);
         context.getCounter(GroupByUserCounters.MAP_CONTEXT_WRITE_IO_EXCEPTION)
            .increment(1);
    } catch (InterruptedException e) {
         LOGGER.error(e);
         context.getCounter(GroupByUserCounters.MAP_CONTEXT_WRITE_INTERRUPTED_EXCEPTION)
            .increment(1);
    }
  
  }
}    


the reduce phase will aggregate counts per URI for each user into a text based map. Note that while Sequence Files are the more optimal way to write this information, we're using Text output to minimize complexity.

user1\t/pages/foo.html:2\t/pages/bar.html:1\t/pages/baz.html:1

/**
 * emit user: uri1:count1...uriN:countN 
 *
 */
public static class GroupViewsByUserReducer extends Reducer {
 
   
 
  /**
   *
   */
  @Override
  public void reduce(Text key, Iterable input, Context context) {
  
    Map URItoCount = new HashMap();
    Iterator it   = input.iterator();
  
    // aggregate views by uri.
    while(it.hasNext()) {
      Text value =  new Text(it.next());
   
      Integer count = URItoCount.get(value);
   
      if(count == null) {
        URItoCount.put(value, Integer.valueOf(1));
      } else {
        URItoCount.put(value, Integer.valueOf(count+1));
      }
   
    }
  
    // emit view:count format, tab delimited.
  
    StringBuffer sbuf = new StringBuffer();
  
    for(Map.Entry entry : URItoCount.entrySet()) {
      sbuf.append(entry.getKey().toString()).append(":").append(entry.getValue());
      sbuf.append("\t");
    }
  
    // convert list to hadoop Text
    String viewBuffer = sbuf.toString();
    LOGGER.debug("reduce: khttps://github.com/arunxarun/cpsamples/blob/master/src/main/java/org/arunxarun/mapreduce/sample/cp/GroupViewsByUser.javay = "+key.toString()+", value = "+viewBuffer);
    Text allViews = new Text(viewBuffer);
  
    try {
      context.write(key, allViews);
    } catch (IOException e) {
      LOGGER.error(e);
      context.getCounter(GroupByUserCounters.REDUCE_CONTEXT_WRITE_IO_EXCEPTION)
         .increment(1);
    } catch (InterruptedException e) {
      LOGGER.error(e);
      context.getCounter(GroupByUserCounters.REDUCE_CONTEXT_WRITE_INTERRUPTED_EXCEPTION)
         .increment(1);
   
    }
  }
}

A Final Note: Unit Testing
I'd like to point out some quick helper classes that  I used to aid my unit testing -- these are at

https://github.com/arunxarun/cpsamples/tree/master/src/main/java/org/arunxarun/mapreduce/unitmocks.

I've started to use these (and/or variants of these) to quickly validate that my mappers and reducers are emitting what I think they should: the bolded lines below allow me to retrieve mapped values and validate them at map time:


        @Test
 public void testMapperValidInput() throws IOException, InterruptedException {

  GroupViewsByUser.GroupViewsByUserMapper mapper = new GroupViewsByUser.GroupViewsByUserMapper();

  MockRecordWriter rw = new MockRecordWriter();
  Mapper.Context context = getMapperContext(mapper,rw);


  LongWritable key = new LongWritable(1);
  Text value = new Text("127.0.0.1\t-\tfrank\t[10/Oct/2012:13:55:36 -0700]\t\"GET /products/prod2.html HTTP/1.0\"\t200\t1002\n");
  mapper.map(key, value, context);
  LongWritable key2 = new LongWritable(2);
  Text value2 = new Text("127.0.0.1\t-\tjoe\t[10/Oct/2012:14:35:22 -0700]\t\"GET /products/prod1.html HTTP/1.0\"\t200\t2326\n");
  mapper.map(key2, value2, context);
  Map map = rw.getMap();
  assertNotNull(map);
  assertTrue(map.size() ==  2);
  assertNotNull(map.get(new Text("joe")));
  assertNotNull(map.get(new Text("frank")));



 }


The getMapperContext() method uses most of these stubs (that return empty values) when it returns a context object that can be passed into the mapper.

        /**
  * return a valid mapper context.
  * 
  * @param mapper
  * @param rw
  * @return
  * @throws IOException
  * @throws InterruptedException
  */
 private Mapper.Context getMapperContext(
   Mapper mapper,
   MockRecordWriter rw) throws IOException,
   InterruptedException {
  Configuration conf = new Configuration();
  TaskAttemptID taskId = new TaskAttemptID();

  Mapper.Context context = mapper.new Context(
    conf, taskId, new MockRecordReader(), rw,
    new MockOutputCommitter(), new MockStatusReporter(),
    new MockInputSplit());

  return context;

 }

The reducer logic is also testable:


        @Test
 public void testReducerValidInput() throws IOException,
   InterruptedException {

  GroupViewsByUser.GroupViewsByUserReducer reducer = 
                  new GroupViewsByUser.GroupViewsByUserReducer();

  Configuration conf = new Configuration();
  TaskAttemptID taskId = new TaskAttemptID("foo", 1, true, 1, 12);
  MockRecordWriter dwr = new MockRecordWriter();

  Reducer.Context context = getReducerContext(reducer,dwr,conf,taskId); 


  Iterable input = new Iterable() {

   @Override
   public Iterator iterator() {
    List list = new ArrayList();

    list.add(new Text("http://foobar.com"));
    list.add(new Text("http://foobar.com"));
    list.add(new Text("http://foobar.com"));
    list.add(new Text("http://foobar.com"));

    return list.iterator();
   }

  };

  Text key = new Text("userkey");

  reducer.reduce(key, input, context);

  Map map = dwr.getMap();

  assertNotNull(map);
  assertTrue(map.size() == 1);
  assertNotNull(map.get(new Text("userkey")));
  Text value = map.get(key);
  assertEquals("http://foobar.com:4\t", value.toString());

 }


again, the getReducerContext() returns a mocked up context that is passed into the reducer and evaluated. The MockRecordWriter can be interrogated for results. Here is that getReducerContext():


  * 
  * return a valid reducer context
  * @param reducer
  * @param dwr
  * @param conf
  * @param id
  * @return
  * @throws IOException
  * @throws InterruptedException
  */
 private Reducer.Context getReducerContext(
   Reducer reducer, 
   MockRecordWriter dwr,
   Configuration conf, 
   TaskAttemptID id) throws IOException, 
   InterruptedException {

  MockOutputCommitter doc = new MockOutputCommitter();
  MockStatusReporter dsr = new MockStatusReporter();
  MockRawKeyValueIterator drkv = new MockRawKeyValueIterator();


  Reducer.Context context = reducer.new Context(
  conf, id, drkv, new MockCounter(), new MockCounter(), dwr,
  doc, dsr, null, Text.class, Text.class);

  return context;
 }

In my next post I'll discuss how to take the output of this job and generate item to item mappings.