title: How to calculate distribution (Histogram) of large amount of data in a distributed system?
body: <p>I am building a metrics reporting system on an instance fleet containing more than 100,000 front-end instances. For any request, every single instance will have a response time. And what I need is the distribution of the response time of every kinds of request over the whole fleet. For example the [Percentile 50, Percentile 90, Percentile 99, Percentile99.9...] of [requestType1, requestType2...requestType1000]. </p>
  
  <p>Every instance will collect the response time take place inside. So over a minute, what one instance collects in memory is the lists of response time of all kinds of requestTypes. For example requestType1 - [1, 2, 3, 4, 1, 2], requestType2 - [2, 2, 3, 2, 1]...... So what I need to do is to process these data and produce the final result. </p>
  
  <p>I tried a lot of designs, my major pain points are the huge size of datapoints I collected of every single requestType, and the expense of communication between instances.  <strong>I will explain my current design below, but I also want to know if there are better designs or some fancy algorithms can aggregate histograms?</strong> </p>
  
  <p>Currently the most promising one is like: Every front-end instance will send their data to a random instance of a mid-layer instance fleet. In this mid-layer fleet, every instance will aggregate all datapoints it gets over a short period of time, e.g. 5 seconds. (It don't have enough memory to hold for a longer time). Then the mid-layer instance will distribute the aggregated data by hash value of requestTypes to back-end instances. What it means is all mid-layer instances will send the datapoints of the same requestTypes to the same back-end instance. Then in the back-end instance I may use a third party's Histogram container (CodaHale's histogram or HdrHistogram) to calculate P50, P90, P99 of  incoming datapoints...The reason I need the mid-layer instance fleet is sending data from front-end instances is expensive, so I want all it's data be sent at once, but not make 100 calls to send to 100 different back-end instances. </p>
  
  <p>The main problem I may think of this design is the relatively high complexity, and if one back-instance is down, I may loss all data of some requestTypes. <strong>So for the system design part, anyone have some better ideas?</strong></p>
  
  <p>The other way I am thinking is to find a fancy algorithm to aggregate existing histograms. The the design above, the data I get will be 100% accurate. But actually I can tolerate some mistakes. For example in CodaHale's histogram and HdrHistogram, I am sure they actually don't save all data points, but applied some advanced math algorithms to get a relatively high precision result with very low cost. And I can use the Histogram library in front-end or mid-layer instances. But the problems is although I can get the [P50, P90, P99...] of every front-end instance or mid-layer instance at a low cost, I couldn't find a way to aggregate them. Because different front-end instance may handle different types of requests, and the distribution of requests to front-end instances are unknown, so simply calculate the average value of ALL P50, P90, P99 will have a lot of inaccuracy. <strong>So does anyone have idea, how can I aggregate multiple CodaHale's histogram or HdrHistogram together? Or are there any algorithms can help to aggregate histograms into one?</strong></p>
  
  <p>========================================================================</p>
  
  <p>I have some new idea last night. Since P50 and P90 are measuring the "average" of all data, I think simple apply weighted average on all P50 and P90 calculated in every mid-layer instance should be good enough. But P99, P99.9 and P99.99 are measuring those outlying data, so an average of P99 of subset may not be accurate. </p>
  
  <p>But if assuming the data in mid-layer instance is relatively random distributed, I can get top 5% of datapoints in every mid-layer instance, and send them to back-end. The 5% of every mid-layer datapoints together is 5% of overall datapoints. And I have more confidence, that the P80 of these 5% data is close to P99 of overall data, P98 of these 5% data is close to P99.9 of overall data, and P99.8 of 5% data is close to P99.99 of overall data. </p>
  
  <p>I hope in this way, I can only transfer 5% of overall data, but get a high accuracy result. What do you think of this way?</p>