/**
 * 将字符串 转换为Node对象  字符串  :0.5 B D
 * @author root
 *
 */
public class Node {

	// 字符串中 第一个元素 初始PR值 为 1
	private double pageRank = 1.0;
	
	// 字符串中 后面的节点列表 数组
	private String[] adjacentNodeNames;

	// PR值 与 数组的分隔符 \t
	public static final char fieldSeparator = '\t';

	public double getPageRank() {
		return pageRank;
	}

	public Node setPageRank(double pageRank) {
		this.pageRank = pageRank;
		return this;
	}

	public String[] getAdjacentNodeNames() {
		return adjacentNodeNames;
	}

	public Node setAdjacentNodeNames(String[] adjacentNodeNames) {
		this.adjacentNodeNames = adjacentNodeNames;
		return this;
	}

	public boolean containsAdjacentNodes() {
		return adjacentNodeNames != null && adjacentNodeNames.length > 0;
	}

	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append(pageRank);

		if (getAdjacentNodeNames() != null) {
			sb.append(fieldSeparator).append(
					StringUtils.join(getAdjacentNodeNames(), fieldSeparator));
		}
		return sb.toString();
	}

	// value =1.0 B D
	public static Node fromMR(String value) throws IOException {
		String[] parts = StringUtils.splitPreserveAllTokens(value, fieldSeparator);
		if (parts.length < 1) {
			throw new IOException("Expected 1 or more parts but received " + parts.length);
		}
		// 1.0 A B D
		Node node = new Node().setPageRank(Double.valueOf(parts[0]));
		if (parts.length > 1) {
			node.setAdjacentNodeNames(Arrays.copyOfRange(parts, 1, parts.length));
		}
		return node;
	}
}


public class RunJob {

	public static enum Mycounter {
		//枚举
		my
	}

	public static void main(String[] args) {
		Configuration config = new Configuration();
		config.set("fs.defaultFS", "hdfs://node01:8020");
		config.set("yarn.resourcemanager.hostname", "node03");
		
//		config.set("mapred.jar", "D://MR/jar/pagerank.jar");
		
		// 收敛值
		double d = 0.001;
		int i = 0;
		while (true) {
			i++;
			try {
				
				// 记录计算的次数
				config.setInt("runCount", i);

				FileSystem fs = FileSystem.get(config);
				Job job = Job.getInstance(config);

				job.setJarByClass(RunJob.class);
				job.setJobName("pr" + i);
				job.setMapperClass(PageRankMapper.class);
				job.setReducerClass(PageRankReducer.class);
				job.setMapOutputKeyClass(Text.class);
				job.setMapOutputValueClass(Text.class);
				job.setInputFormatClass(KeyValueTextInputFormat.class);

				Path inputPath = new Path("/MR/pagerank/input/pagerank.txt");

				if (i > 1) {
					inputPath = new Path("/MR/pagerank/output/pr" + (i - 1));
				}
				FileInputFormat.addInputPath(job, inputPath);

				Path outpath = new Path("/MR/pagerank/output/pr" + i);
				if (fs.exists(outpath)) {
					fs.delete(outpath, true);
				}
				FileOutputFormat.setOutputPath(job, outpath);

				boolean f = job.waitForCompletion(true);
				if (f) {
					System.out.println("success.");
					// 获取 计数器 中的差值
					long sum = job.getCounters().findCounter(Mycounter.my).getValue();

					System.out.println("SUM:  " + sum);
					double avgd = sum / 4000.0;
					if (avgd < d) {
						break;
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	static class PageRankMapper extends Mapper<Text, Text, Text, Text> {
		@Override
		protected void map(Text key, Text value, Context context)
				throws IOException, InterruptedException {
			int runCount = context.getConfiguration().getInt("runCount", 1);
			// A	B	D
			String page = key.toString();
			Node node = null;
			if (runCount == 1) { //第一次计算 初始化PR值为1.0
				node = Node.fromMR("1.0" + "\t" + value.toString());
			} else {
				node = Node.fromMR(value.toString());
			}
			// A:1.0 B D
			// 将计算前的数据输出  reduce计算做差值
			context.write(new Text(page), new Text(node.toString()));

			if (node.containsAdjacentNodes()) {
				double outValue = node.getPageRank() / node.getAdjacentNodeNames().length;
				for (int i = 0; i < node.getAdjacentNodeNames().length; i++) {
					String outPage = node.getAdjacentNodeNames()[i];
					// B:0.5
					// D:0.5
					context.write(new Text(outPage), new Text(outValue + ""));
				}
			}
		}
	}

	static class PageRankReducer extends Reducer<Text, Text, Text, Text> {
		@Override
		protected void reduce(Text key, Iterable<Text> iterable, Context context)
				throws IOException, InterruptedException {
			double sum = 0.0;
			Node sourceNode = null;
			for (Text i : iterable) {
				Node node = Node.fromMR(i.toString());
				//A:1.0 B D
				if (node.containsAdjacentNodes()) {
					// 计算前的数据 // A:1.0 B D
					sourceNode = node;
				} else {
					// B:0.5 // D:0.5
					sum = sum + node.getPageRank();
				}
			}

			// 计算新的PR值  4为页面总数
			double newPR = (0.15 / 4.0) + (0.85 * sum);
			System.out.println("*********** new pageRank value is " + newPR);

			// 把新的pr值和计算之前的pr比较
			double d = newPR - sourceNode.getPageRank();

			int j = (int) (d * 1000.0);
			j = Math.abs(j);
			System.out.println(j + "___________");
			// 累加
			context.getCounter(Mycounter.my).increment(j);

			sourceNode.setPageRank(newPR);
			context.write(key, new Text(sourceNode.toString()));
		}
	}
}