public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();

		// 生成公用的随机向量文件
		RandomGeneration.randomVector(conf);

		FileSystem hdfs = FileSystem.get(conf);
		hdfs.delete(outputFolder, true);
		hdfs.delete(outputFolder2, true);

		// 获取分布式缓存文件路径
		DistributedCache.addCacheFile(randomVectorPath.toUri(), conf);

		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		if (otherArgs.length != 0) {
			System.err.println("Usage: 神马都没有！");
			System.exit(2);
		}
		Job job = new Job(conf, "HSEM" + "_" + String.valueOf(RANDOM_VECTORS)
				+ "_" + String.valueOf(PERMUTATION) + "_"
				+ String.valueOf(WINDOW));
		job.setJarByClass(EntityDriver.class);

		job.setMapperClass(EntityMap.class);
		job.setReducerClass(EntityReduce.class);

		if (PERMUTATION >= 11) // 依据随机变换数量决定reduce数量
			job.setNumReduceTasks(11);
		else
			job.setNumReduceTasks(PERMUTATION);

		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(EntityData.class);

		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(EntityPairData.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		FileStatus stats[] = hdfs.listStatus(inputFolder);
		for (int i = 0; i < stats.length; i++) {
			if (!stats[i].isDir())
				FileInputFormat.addInputPath(job, stats[i].getPath());
		}

		 SequenceFileOutputFormat.setOutputCompressionType(job,
		 CompressionType.RECORD);
		 SequenceFileOutputFormat.setOutputCompressorClass(job,
		 GzipCodec.class);
		SequenceFileOutputFormat.setOutputPath(job, outputFolder);
		// FileOutputFormat.setOutputPath(job, outputFolder);
		if (job.waitForCompletion(true)) {
			Job secondJob = new Job(conf, "Deduplication" + "_"
					+ String.valueOf(RANDOM_VECTORS) + "_"
					+ String.valueOf(PERMUTATION) + "_"
					+ String.valueOf(WINDOW));
			secondJob.setJarByClass(EntityDriver.class);
			secondJob.setInputFormatClass(SequenceFileInputFormat.class);
			secondJob.setMapperClass(DeduplicationMap.class);
			secondJob.setReducerClass(DeduplicationReduce.class);
			secondJob.setMapOutputKeyClass(Text.class);
			secondJob.setMapOutputValueClass(EntityPairData.class);
			secondJob.setNumReduceTasks(11);
			secondJob.setOutputKeyClass(Text.class);
			secondJob.setOutputValueClass(FloatWritable.class);

			FileStatus stats2[] = hdfs.listStatus(outputFolder);
			for (int i = 0; i < stats2.length; i++) {
				if (!stats2[i].isDir())
					FileInputFormat
							.addInputPath(secondJob, stats2[i].getPath());
				// SequenceFileInputFormat
				// .addInputPath(secondJob, stats2[i].getPath());
			}

			FileOutputFormat.setOutputPath(secondJob, outputFolder2);
			System.exit(secondJob.waitForCompletion(true) ? 0 : 1);
		}
		// System.exit(job.waitForCompletion(true) ? 0 : 1);
	}