2

SQLGroupByに似たようなコードを作成しました。

私が取ったデータセットはここにあります：

250788681419,20090906,200937,200909,619、SUNDAY、WEEKEND、ON-NET、MORNING、OUTGOING、VOICE、25078、PAY_AS_YOU_GO_PER_SECOND_PSB、SUCCESSFUL-RELEASEDBYSERVICE、17,0,1,21.25,635-10-112-30455

public class MyMap extends Mapper<LongWritable, Text, Text, DoubleWritable> {

public void map(LongWritable key, Text value, Context context) throws IOException 
{

        String line = value.toString();
        String[] attribute=line.split(",");
        double rs=Double.parseDouble(attribute[17]);

        String comb=new String();
        comb=attribute[5].concat(attribute[8].concat(attribute[10]));

            context.write(new Text(comb),new DoubleWritable (rs));

    }
 } 
public class MyReduce extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {

protected void reduce(Text key, Iterator<DoubleWritable> values, Context context) 
          throws IOException, InterruptedException {

             double sum = 0;
             Iterator<DoubleWritable> iter=values.iterator();
                while (iter.hasNext()) 
                {
                    double val=iter.next().get();
                    sum = sum+ val;
                }
                context.write(key, new DoubleWritable(sum));
        };
     }

マッパーでは、その値として17番目の引数をレデューサーに送信して合計します。ここで、14番目の引数も合計したいのですが、それをレデューサーに送信するにはどうすればよいですか？

4

1 に答える 1

2

データ型が同じである場合は、ArrayWritableクラスの作成がこれで機能するはずです。クラスは次のようになります。

public class DblArrayWritable extends ArrayWritable 
{ 
    public DblArrayWritable() 
    { 
        super(DoubleWritable.class); 
    }
}

マッパークラスは次のようになります。

public class MyMap extends Mapper<LongWritable, Text, Text, DblArrayWritable> 
{
  public void map(LongWritable key, Text value, Context context) throws IOException 
  {

    String line = value.toString();
    String[] attribute=line.split(",");
    DoubleWritable[] values = new DoubleWritable[2];
    values[0] = Double.parseDouble(attribute[14]);
    values[1] = Double.parseDouble(attribute[17]);

    String comb=new String();
    comb=attribute[5].concat(attribute[8].concat(attribute[10]));

    context.write(new Text(comb),new DblArrayWritable.set(values));

  }
}

これで、レデューサーでDblArrayWritableの値を反復処理できるようになります。

ただし、サンプルデータに基づくと、それらは別々のタイプである可能性があります。トリックを実行するObjectArrayWritableクラスを実装できる可能性がありますが、これについては確信が持てず、サポートするものがあまりありません。それが機能する場合、クラスは次のようになります。

public class ObjArrayWritable extends ArrayWritable 
{ 
    public ObjArrayWritable() 
    { 
        super(Object.class); 
    }
}

これを処理するには、値を連結し、テキストとしてレデューサーに渡します。レデューサーは値を再度分割します。

もう1つのオプションは、独自のWritableクラスを実装することです。これがどのように機能するかのサンプルです：

public static class PairWritable implements Writable 
{
   private Double myDouble;
   private String myString;

    // TODO :-  Override the Hadoop serialization/Writable interface methods
    @Override
    public void readFields(DataInput in) throws IOException {
            myLong = in.readDouble();
            myString = in.readUTF();
    }

    @Override
    public void write(DataOutput out) throws IOException {
            out.writeDouble(myLong);
            out.writeUTF(myString);
    }

    //End of Implementation

    //Getter and Setter methods for myLong and mySring variables
    public void set(Double d, String s) {
        myDouble = d;
        myString = s;
    }

    public Long getLong() {
        return myDouble;
    }
    public String getString() {
        return myString;
    }

}
于 2013-01-25T06:23:16.573 に答える