基于Hadoop大数据平台实现遗传算法并行化

package graduatedesign.GA; import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.DoubleWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Mapper.Context; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; public class GAdesign extends Configured implements Tool { //继承类 /* * 驱动函数 */ public int run(String[] args) throws Exception{ Configuration conf = getConf(); //获取Configuration实例 Job job = new Job(conf); //创建任务实例 job.setJarByClass(getClass()); //创建工作类 FileSystem fs = FileSystem.get(conf); //获取环境变量 fs.delete(new Path("/user/hadoop/output"),true); //删除已存在输出路径 job.setMapperClass(GAmapper.class); //Mapper工作类 job.setInputFormatClass(NLineInputFormat.class); //设置多行类 job.getConfiguration().set(NLineInputFormat.LINES_PER_MAP,"20000");//指定每20000行为一个splits job.setOutputKeyClass(Text.class); //设置输出类型为文本类型 job.setOutputValueClass(NullWritable.class); //不输出值 FileInputFormat.addInputPath(job, new Path("/user/hadoop/input/part-r-00000")); //设置输入文件路径 FileOutputFormat.setOutputPath(job, new Path("/user/hadoop/output")); //设置输出文件路径 job.waitForCompletion(true); //启动任务 return 0; } /* * 对输出文件重命名 */ static void testRename() throws Exception { String hdfsPath = "/user/hadoop/"; //需要重命名文件所在文件夹 String newhdfs = "/user/hadoop/output/_SUCCESS"; //需要删除的输出文件 String inputsc = "/user/hadoop/input"; //需要输出的上次输入文件 Configuration conf = new Configuration(); FileSystem hdfs = FileSystem.get(conf); //获取系统设置 // Path dst = new Path("hdfs://localhost:9000/user/"); hdfs.delete(new Path(newhdfs), true); //删除输出的无用文件 hdfs.delete(new Path(inputsc), true); //删除上次的输入文件 Path frpath = new Path(hdfsPath + "output"); //待命名文件 Path topath = new Path(hdfsPath + "input"); //目的文件名 hdfs.rename(frpath, topath); //重命名文件 } public static void main(String[] args) throws Exception { Tool tool = new GAdesign(); for(int i=0;i<2;i++){ //设置迭代次数 ToolRunner.run(tool, args); //运行mapreduce testRename(); //重命名设置 } /* * */ Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(conf); fs.delete(new Path("/user/hadoop/output"),true); Job job = new Job(conf,"GAdesign"); job.setJarByClass(GAdesign.class); job.setMapperClass(MyMapper.class); job.setReducerClass(MyReducer.class); job.setOutputKeyClass(DoubleWritable.class); job.setOutputValueClass(Text.class); job.setCombinerClass(MyReducer.class); FileInputFormat.addInputPath(job, new Path("/user/hadoop/input/part-r-00000")); FileOutputFormat.setOutputPath(job, new Path("/user/hadoop/output")); System.exit(job.waitForCompletion(true) ? 0:1); /* * */ } /* * Mapper类，负责惊醒交叉、变异、选择运算 */ public static class GAmapper extends Mapper<Object,Text,Text,NullWritable>{ int t = 0; String[] individuals = new String[20000]; //定义种群的规模 private Text word = new Text(); //写入下一代个体 public void map(Object key,Text value,Context context) throws IOException, InterruptedException{ StringTokenizer itr = new StringTokenizer(value.toString()); //读取文件 /* * 将文件转化为数组，方便进行下一步处理 */ while(itr.hasMoreTokens()){ individuals[t%20000]=itr.nextToken(); } t++; if(t%20000==0){ CrossOver.crossover(individuals); //进行交叉操作 CrossOver.mutation(individuals); //变异操作 Rouletee.rouletee(individuals); //选择操作 /* * 输出文件，记录每代的最大值 */ for(int i=0;i<individuals.length;i++){ word.set(individuals[i]); //类型转化，String转换成Text context.write(word, NullWritable.get());//写入Mapperout } } } } }