java/pig-galago/src/com/yahoo/pig/builtin/GFCross.java

   1 /*
   2  * Copyright (c) 2007 Yahoo! Inc.  All rights reserved.
   3  * See accompanying LICENSE file.
   4  */
   5 package com.yahoo.pig.builtin;
   6
   7 import java.io.IOException;
   8 import java.util.Random;
   9
  10 import com.yahoo.pig.GroupFunc;
  11 import com.yahoo.pig.data.Datum;
  12 import com.yahoo.pig.data.Tuple;
  13
  14 public class GFCross extends GroupFunc {
  15         int numInputs, myNumber, numGroupsPerInput;
  16
  17         public static int DEFAULT_PARALLELISM = 96;
  18         @Override
  19         public Datum[] exec(Tuple input) {
  20
  21                 try{
  22                         numInputs = Integer.parseInt(input.getAtomField(0).strval());
  23                         myNumber = Integer.parseInt(input.getAtomField(1).strval());
  24
  25
  26                         numGroupsPerInput = (int) Math.ceil(Math.pow(DEFAULT_PARALLELISM, 1.0/numInputs));
  27                         int numGroupsGoingTo = (int) Math.pow(numGroupsPerInput,numInputs - 1);
  28
  29                         int[] digits = new int[numInputs];
  30                         for (int i=0; i<numInputs; i++){
  31                                 if (i == myNumber){
  32                                         Random r = new Random(System.currentTimeMillis());
  33                                         digits[i] = r.nextInt(numGroupsPerInput);
  34                                 }else{
  35                                         digits[i] = 0;
  36                                 }
  37                         }
  38
  39                         Tuple[] groups = new Tuple[numGroupsGoingTo];
  40                         for (int i=0; i<numGroupsGoingTo; i++){
  41                                 groups[i] = toTuple(digits);
  42                                 next(digits);
  43                         }
  44                         return groups;
  45
  46                 }catch(IOException e){
  47                         RuntimeException re = new RuntimeException(e.getMessage());
  48                         re.setStackTrace(e.getStackTrace());
  49                         throw re;
  50                 }
  51         }
  52
  53         private Tuple toTuple(int[] digits) throws IOException{
  54                 Tuple t = new Tuple(numInputs);
  55                 for (int i=0; i<numInputs; i++){
  56                         t.setField(i, digits[i]);
  57                 }
  58                 return t;
  59         }
  60
  61         private void next(int[] digits){
  62                 for (int i=0; i<numInputs; i++){
  63                         if (i== myNumber)
  64                                 continue;
  65                         else{
  66                                 if (digits[i] == numGroupsPerInput - 1){
  67                                         digits[i] = 0;
  68                                 }else{
  69                                         digits[i]++;
  70                                         break;
  71                                 }
  72                         }
  73                 }
  74         }
  75
  76 }