Method exec

src/org/apache/pig/builtin/COR.java:189–243

(Tuple input)

        }

        @Override
        public DataBag exec(Tuple input) throws IOException {
            if (input == null || input.size() == 0)
                return null;

            Tuple combined;    
            try{
                combined = combine((DataBag)input.get(0));    
            }catch(Exception e){
                throw new IOException("Caught exception in COR.Final", e);
            }
            int count = 0;
            //for each pair of input schema combined contain 2 member. first member
            //is tuple containing sum_x,sum_y,sum_x_y, sum_x_square, sum_y_square
            //and second is number of tuples from which it is calculated . 
            //So if arity of combined is n then total number of schemas would be 
            //root of x*x - x - n =0 
            int totalSchemas=2;
            while(totalSchemas*(totalSchemas-1)<combined.size()){
                totalSchemas++;
            } 
                
            DataBag output = DefaultBagFactory.getInstance().newDefaultBag();
            for(int i=0;i<totalSchemas;i++){
                for(int j=i+1;j<totalSchemas;j++){
                    Tuple result = TupleFactory.getInstance().newTuple(3);
                    try{
                        if(flag){
                            result.set(0, schemaName.elementAt(i));
                            result.set(1, schemaName.elementAt(j));
                        } 
                        else{
                            result.set(0, "var"+i);
                            result.set(1, "var"+j);
                        }
                        Tuple tup = (Tuple)combined.get(count);
                        double tempCount = (Long)combined.get(count+1);
                        double sum_x_y = (Double)tup.get(0);
                        double sum_x = (Double)tup.get(1);
                        double sum_y = (Double)tup.get(2);
                        double sum_x_square = (Double)tup.get(3);
                        double sum_y_square = (Double)tup.get(4);
                        double correl = (tempCount*sum_x_y - sum_x*sum_y)/Math.sqrt((tempCount*sum_x_square-sum_x*sum_x)*(tempCount*sum_y_square-sum_y*sum_y));
                        result.set(2, correl);
                    }catch(Exception e){
                        System.err.println("Failed to process input record; error - " + e.getMessage());
                        return null;
                    } 
                    output.add(result);
                    count+=2;
                }
            }

            return output;
        }
    }

    /**