Beijing Red Star

Shenzhen Thunder

Guangzhou Honda

Beijing Rising

Guangzhou Development Bank

Tencent

Back of Beijing    

    Beijing

    Guangzhou

    Shenzhen

    Xian

package com;

import java.io.IOException;

import java.util.ArrayList;

import java.util.Iterator;

import java.util.List;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Test {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration config = new  Configuration();

        config.set("fs.defaultFS", "hdfs://192.168.0.100:9000");

        config.set("yarn.resourcemanager.hostname", "192.168.0.100");

        FileSystem fs = FileSystem.get(config);

        Job job = Job.getInstance(config);

        job.setJarByClass(Test.class);

        //设置所用到的map类

        job.setMapperClass(myMapper.class);

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(Text.class);

        //设置用到的reducer类

        job.setReducerClass(myReducer.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(Text.class);

        //设置s输入输出地址

        FileInputFormat.addInputPath(job, new Path("/FactoryName/"));

        Path path = new Path("/output2/");

        if(fs.exists(path)){

            fs.delete(path, true);

        }

        //指定文件的输出地址

        FileOutputFormat.setOutputPath(job, path);

        //启动处理任务job

        boolean completion = job.waitForCompletion(true);

        if(completion){

            System.out.println("Job Success!");

        }

    }

    public static class myMapper extends Mapper<Object, Text, Text, Text> {

        // 实现map函数

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

            String temp=new String();// 左右表标识

            String values=value.toString();

            String words[]=values.split("\t");

            String mapkey = new String();

            String mapvalue = new String(); 

            //右表：1  Beijing

            if (words[].charAt() >= '' && words[].charAt() <= '') {

                mapkey = words[];

                mapvalue =words[];

                temp = "";

            }else{

                //左表：Beijing Red Star  1

                mapkey = words[];

                mapvalue =words[];

                temp = "";

            }

            // 输出左右表

            //左表：(1,1+Beijing Red Star)

            //右表：(1,2+Beijing)

            context.write(new Text(mapkey), new Text(temp + "+"+ mapvalue));

        }

    }

    //reduce解析map输出，将value中数据按照左右表分别保存

    public static class myReducer extends Reducer<Text, Text, Text, Text> {

        // 实现reduce函数

        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            List<String> factory = new ArrayList<String>();

            List<String> address = new ArrayList<String>();

            for (Text value : values) {

                // 取得左右表标识

                char temp=(char) value.charAt();

                String words[]=value.toString().split("[+]");//1,Beijing Red Star

                if(temp==''){

                    factory.add(words[]);// 左表

                }

                if(temp==''){

                    address.add(words[]);// 右表

                }

            }

            //求出笛卡尔积，并输出

            for (String f : factory) {

                for (String a : address) {

                    context.write(new Text(f), new Text(a));

                }

            }

        }

    }

}