java - 如何将我的 MapReduce 作业的结果存储在 Hashmap 中并按值排序？

Question

我是 HBase MapReduce 工作的新手，我想计算我表上的前 10 个用户。

在我的 Reducer 类中，我放置了一个本地哈希图来存储每个结果排序图。

我的问题是:

由于添加 'System.out.println' 语句不起作用，如何打印出我的 hashmap 的内容？

公共(public)类 MyScanner2 {

static Configuration conf; 
static long startTimestamp;
static long stopTimestamp;
static Scan myScan;
static String tableToScan = "VStable";

public static void main(String[] args) throws IOException, ParseException, InterruptedException, ClassNotFoundException {
    // TODO Auto-generated method stub

    initScanner();

    @SuppressWarnings("deprecation")
    Job job = new Job(conf, "TOP10_users"); //TOP10_users is the name of the job
    job.setJarByClass(MyScanner2.class);
    FileOutputFormat.setOutputPath(job, new Path("hdfs://zwinf5q45:8020/user/hdfs/top10users"));
    TableMapReduceUtil.initTableMapperJob(Bytes.toBytes(tableToScan), myScan, Mapper1.class, ImmutableBytesWritable.class, IntWritable.class, job);
    TableMapReduceUtil.initTableReducerJob("stats", Reducer1.class, job);
    //System.out.println(MyReducer.getMap().toString());
    System.exit(job.waitForCompletion(true) ? 0 : 1);

}

public static void initScanner() throws IOException, ParseException{

    conf = HBaseConfiguration.create();
    conf.set("hbase.rootdir", "hdfs://zwinf5q45:8020/apps/hbase/data");
    conf.set("hbase.zookeeper.quorum", "zwinf5q46,zwinf5q44,zwinf5q43,zwinf5q42,zwinf5q41");
    conf.set("zookeeper.znode.parent", "/hbase-unsecure");

    startTimestamp = convertToTimestamp("2014-05-21");
    stopTimestamp = convertToTimestamp("2014-05-22");;

    myScan = new Scan();
    myScan.setStartRow(Bytes.toBytes(startTimestamp));
    myScan.setStopRow(Bytes.toBytes(stopTimestamp));
    myScan.addColumn(Bytes.toBytes("infos"), Bytes.toBytes("bucketID"));
    myScan.setCaching(1000);



}

 public static long convertToTimestamp(String str_date) throws ParseException{

     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
     java.util.Date date = sdf.parse(str_date);
     java.sql.Timestamp timestamp= new java.sql.Timestamp(date.getTime());

     return timestamp.getTime();
}

}

类 Mapper1 扩展 TableMapper {

private int numRecords = 0;
private static final IntWritable one = new IntWritable(1);

@Override
public void map(ImmutableBytesWritable row, Result values, Context context) throws IOException {

    // extract resource
    if (values.isEmpty()){
        System.out.println("The scanner is empty");
    }
    else{

        ImmutableBytesWritable resource = new ImmutableBytesWritable(values.getValue(Bytes.toBytes("infos"), Bytes.toBytes("bucketID")));

        try {
            context.write(resource, one);
        } catch (InterruptedException e) {
            throw new IOException(e);
        }
        numRecords++;
        if ((numRecords % 10000) == 0) {
            context.setStatus("mapper processed " + numRecords + " records so far");
        }
    }


}

}

类 Reducer1 扩展 TableReducer {

static HashMap<String,Integer> map = new HashMap<String,Integer>();

public void reduce(ImmutableBytesWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

    int sum = 0;

    for (IntWritable val : values) {
        sum += val.get();
    }        
    map.put(key.toString(),sum);
    System.out.println ("HashMap content" + Arrays.toString (map.values().toArray ()));

}

}

Answer 1

至于为什么你的输出是错误的试试

System.out.println (Arrays.toString (map.values().toArray ()));