public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        double pr = 0.1;
        double each_leak = Double.parseDouble(context.getConfiguration().get("each_leak"));
        pr += each_leak;
        double old_pr = -1000000.0;
        String toIds = "";
        for (Text val : values) {
            String s = val.toString();
            String[] args = s.split("\\$");
            if (s.startsWith("0")) {
                if (args.length >= 3) {
                    toIds = args[2];
                }
                old_pr = Double.parseDouble(args[1]);
            } else {
                double addpr = Double.parseDouble(args[1]);
                pr += addpr;
            }
        }
        context.write(key, new Text(String.valueOf(pr) + "|" + String.valueOf(old_pr) + "$" + toIds));
    }