Hadoop HelloWorld Examples - 求k临近点(+自定义变量+参数传入)(三)
cpoint.x), (2.0f)) +
Math.pow((value.y - cpoint.y), (2.0f)));
context.write(new FloatWritable(dis), value);
}
}
public static class KNPointReducer extends Reducer
{
@Override
public void reduce(FloatWritable key, Iterable values, Context context) throws IOException, InterruptedException
{
for(Point2D val : values)
{
context.write(key, val);
}
}
}
public static void main(String[] args) throws Exception
{
Configuration conf = new Configuration();
conf.addResource(new Path("/usr/local/hadoop/conf/core-site.xml"));
conf.set("cPoint", "0 0");//Define the center point. We calculate the others' distance with the center point.
Job job = new Job(conf);
job.setInputFormatClass(KNPointInputFormat.class);//My own input format
job.setOutputFormatClass(TextOutputFormat.class);
job.setJarByClass(KNPoints.class);
job.setMapperClass(KNPointMapper.class);
job.setReducerClass(KNPointReducer.class);
job.setOutputKeyClass(FloatWritable.class);
job.setOutputValueClass(Point2D.class);
String in = "hdfs://localhost:9000/user/hadoop/input/data";
String out = "hdfs://localhost:9000/user/hadoop/output";
FileSystem fs = FileSystem.get(conf);
fs.copyFromLocalFile(new Path("/home/hadoop/CodeSpace/KNPoints/data"),
new Path("hdfs://localhost:9000/user/hadoop/input/"));
FileInputFormat.addInputPath(job, new Path(in));
FileOutputFormat.setOutputPath(job, new Path(out));
if(fs.exists(new Path(out)) == true)
{
fs.delete(new Path(out),true);
}
job.waitForCompletion(true);
}