MapReduce joinケース

2つのファイルdept.txt

10,ACCOUNTING,NEW YORK
20,RESEARCH,DALLAS
30,SALES,CHICAGO
40,OPERATIONS,BOSTON

フィールド解釈:部門番号、部門名、所在都市emp.txt

7369,SMITH,CLERK,7902,1980-12-17,800.00,,20
7499,ALLEN,SALESMAN,7698,1981-2-20,1600.00,300.00,30
7521,WARD,SALESMAN,7698,1981-2-22,1250.00,500.00,30
7566,JONES,MANAGER,7839,1981-4-2,2975.00,,20
7654,MARTIN,SALESMAN,7698,1981-9-28,1250.00,1400.00,30
7698,BLAKE,MANAGER,7839,1981-5-1,2850.00,,30
7782,CLARK,MANAGER,7839,1981-6-9,2450.00,,10
7788,SCOTT,ANALYST,7566,1987-4-19,3000.00,,20
7839,KING,PRESIDENT,,1981-11-17,5000.00,,10
7844,TURNER,SALESMAN,7698,1981-9-8,1500.00,0.00,30
7876,ADAMS,CLERK,7788,1987-5-23,1100.00,,20
7900,JAMES,CLERK,7698,1981-12-3,950.00,,30
7902,FORD,ANALYST,7566,1981-12-3,3000.00,,20
7934,MILLER,CLERK,7782,1982-1-23,1300.00,,10
8888,OTHER,PROGRAM,7839,1988-1-23,10300.00,,

フィールドの説明:従業員番号、従業員名、...部門番号
SQLを書くなら、かなりのeasy

select a.deptno,a.deptname,b.empno,b.empname from dept a,emp b where a.deptno = b.deptno

ソース:

public class EmpInfo implements Writable {

    private int deptno;
    private String deptName;
    private int empno;
    private String empName;
    private int flag;
   
   // getter setter    ，    ...
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(deptno);
        out.writeUTF(deptName);
        out.writeInt(empno);
        out.writeUTF(empName);
        out.writeInt(flag);
    }

    @Override
    public void readFields(DataInput in) throws IOException {

        this.deptno = in.readInt();
        this.deptName = in.readUTF();
        this.empno = in.readInt();
        this.empName = in.readUTF();
        this.flag = in.readInt();
    }
}

Mapperエンド:

public static class MrJoinMapper extends Mapper<LongWritable, Text, IntWritable, EmpInfo>{
        String filename;
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            //           
            FileSplit split = (FileSplit)context.getInputSplit();
            filename = split.getPath().getName();
        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] split = value.toString().split(",");
            int deptNo;
            if (filename.equals("dept.txt")){
                EmpInfo empInfo = new EmpInfo();
                deptNo = Integer.parseInt(split[0].trim());
                empInfo.setDeptno(deptNo);
                empInfo.setDeptName(split[1]);
                empInfo.setEmpName("");
                empInfo.setEmpno(0);
                empInfo.setFlag(1);
                context.write(new IntWritable(deptNo),empInfo);
            }else {
                if (split.length > 6){
                    EmpInfo empInfo = new EmpInfo();
                    deptNo = Integer.parseInt(split[7].trim());
                    empInfo.setEmpno(Integer.parseInt(split[0].trim()));
                    empInfo.setEmpName(split[1]);
                    empInfo.setDeptno(deptNo);
                    empInfo.setDeptName("");
                    empInfo.setFlag(2);
                    System.out.println("empInfo :"+empInfo);
                    context.write(new IntWritable(deptNo),empInfo);
                }
            }
        }
    }

Reducer端子

public static class MrJoinReducer extends Reducer<IntWritable, EmpInfo,EmpInfo,NullWritable>{        
        @Override
        protected void reduce(IntWritable key, Iterable<EmpInfo> values, Context context) throws IOException, InterruptedException {
            List<EmpInfo> list = new ArrayList<>(20);
            String deptName = "";
            for (EmpInfo emp : values) {
                int flag = emp.getFlag();
                if (flag == 2){
                    EmpInfo temp = new EmpInfo();//       new  ，   ...(   )
                    temp.setEmpno(emp.getEmpno());
                    temp.setEmpName(emp.getEmpName());
                    temp.setDeptno(emp.getDeptno());
                    list.add(temp);
                }else {
                    deptName = emp.getDeptName();
                }
            }
            for (EmpInfo info : list) {
                info.setDeptName(deptName);
                context.write(info, NullWritable.get());
            }
        }
    }

出力効果:

10, deptName='ACCOUNTING', empno=7934, empName='MILLER
10, deptName='ACCOUNTING', empno=7839, empName='KING
10, deptName='ACCOUNTING', empno=7782, empName='CLARK
20, deptName='RESEARCH', empno=7876, empName='ADAMS
20, deptName='RESEARCH', empno=7788, empName='SCOTT
20, deptName='RESEARCH', empno=7369, empName='SMITH
20, deptName='RESEARCH', empno=7566, empName='JONES
20, deptName='RESEARCH', empno=7902, empName='FORD
30, deptName='SALES', empno=7844, empName='TURNER
30, deptName='SALES', empno=7499, empName='ALLEN
30, deptName='SALES', empno=7698, empName='BLAKE
30, deptName='SALES', empno=7654, empName='MARTIN
30, deptName='SALES', empno=7521, empName='WARD
30, deptName='SALES', empno=7900, empName='JAMES

まとめ:1.2つのファイルjoinは、そのデータがどのテーブルに由来するかを知ることが重要であるため、flagを設定してデータがどのテーブルに由来するかを判断しなければならない.map側はsetup法により対応するファイル名③を得る.joinのkeyはmapperの出力keyです