combine results of different data chunk @param values DataBag containing partial results computed on different data chunks @return output Tuple containing combined data @throws IOException
(DataBag values)
| 250 | * @throws IOException |
| 251 | */ |
| 252 | static protected Tuple combine(DataBag values) throws IOException { |
| 253 | Tuple output = TupleFactory.getInstance().newTuple(); |
| 254 | Tuple tuple; // copy of DataBag values |
| 255 | tuple = TupleFactory.getInstance().newTuple(); |
| 256 | |
| 257 | try{ |
| 258 | for (Iterator<Tuple> it = values.iterator(); it.hasNext();) { |
| 259 | Tuple t = it.next(); |
| 260 | tuple.append(t); |
| 261 | } |
| 262 | }catch(Exception e){} |
| 263 | |
| 264 | try{ |
| 265 | int size = ((Tuple)tuple.get(0)).size(); |
| 266 | for(int i=0;i<size;i=i+2){ |
| 267 | long count = 0; |
| 268 | double sum_x_y = 0.0; |
| 269 | double sum_x = 0.0; |
| 270 | double sum_y = 0.0; |
| 271 | double sum_x_square = 0.0; |
| 272 | double sum_y_square = 0.0; |
| 273 | for(int j=0;j<tuple.size();j++){ |
| 274 | Tuple temp = (Tuple)tuple.get(j); |
| 275 | Tuple tem = (Tuple)temp.get(i); |
| 276 | count += (Long)temp.get(i+1); |
| 277 | sum_x_y += (Double)tem.get(0); |
| 278 | sum_x += (Double)tem.get(1); |
| 279 | sum_y += (Double)tem.get(2); |
| 280 | sum_x_square += (Double)tem.get(3); |
| 281 | sum_y_square += (Double)tem.get(4); |
| 282 | } |
| 283 | Tuple result = TupleFactory.getInstance().newTuple(5); |
| 284 | result.set(0, sum_x_y); |
| 285 | result.set(1, sum_x); |
| 286 | result.set(2, sum_y); |
| 287 | result.set(3, sum_x_square); |
| 288 | result.set(4, sum_y_square); |
| 289 | output.append(result); |
| 290 | output.append(count); |
| 291 | } |
| 292 | }catch(Exception e){ |
| 293 | throw new IOException("Caught exception in COR.combine", e); |
| 294 | } |
| 295 | |
| 296 | return output; |
| 297 | } |
| 298 | |
| 299 | /** |
| 300 | * compute sum(XY), sum(X), sum(Y), sum(XX), sum(YY) from given data sets |