項目github位址：bitcarmanlee easy-algorithm-interview-and-practice
歡迎大家star，留言，一起學習進步
在MR中，類似于join類的操作非常常見。在關系型資料庫中，join就是最強大的功能之一。在hive中，jion操作也十分常見。現在，本部落客就手把手教會大家怎麼在MR中實作join操作。為了友善起見，本文就以left join為視角來實作。
1.資料準備
關于什麼是join，什麼是left join，本文就先不讨論了。先準備如下資料：
cat employee.txt
jd,david
jd,mike
tb,mike
tb,lucifer
elong,xiaoming
elong,ali
tengxun,xiaoming
tengxun,lilei
xxx,aaa
cat salary.txt
jd,1600
tb,1800
elong,2000
tengxun,2200
然後将兩個檔案分别put到hdfs上面
hadoop fs -put employee.txt /tmp/wanglei/employee/employee.txt
hadoop fs -put salary.txt /tmp/wanglei/salary/salary.txt
我們想用employee左關聯salary。至此，資料準備已經完畢。
2.建立一個maven項目
建立一個maven項目，pom檔案如下：
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>leilei.bit.edu</groupId>
<artifactId>testjoin</artifactId>
<version>1.0</version>
<packaging>jar</packaging>
<name>testjoin</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<jarfile.name>testjoin</jarfile.name>
<jar.out.dir>jar</jar.out.dir>
</properties>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>2.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.mrunit</groupId>
<artifactId>mrunit</artifactId>
<version>1.0.0</version>
<classifier>hadoop2</classifier>
</dependency>
<dependency>
<groupId>org.anarres.lzo</groupId>
<artifactId>lzo-hadoop</artifactId>
<version>1.0.2</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<version>2.4</version>
<configuration>
<finalName>${jarfile.name}</finalName>
<outputDirectory>${jar.out.dir}</outputDirectory>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>1.6</source>
<target>1.6</target>
<encoding>UTF-8</encoding>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
</plugin>
</plugins>
</build>
</project>
然後開始實作left join的邏輯：
package leilei.bit.edu.testjoin;
import java.io.IOException;
import java.util.Vector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class LeftJoin extends Configured implements Tool{
public static final String DELIMITER = ",";
public static class LeftJoinMapper extends
Mapper<LongWritable,Text,Text,Text> {
protected void map(LongWritable key, Text value, Context context)
throws IOException,InterruptedException {
/*
* 拿到兩個不同檔案，區分出到底是哪個檔案，然後分别輸出
*/
String filepath = ((FileSplit)context.getInputSplit()).getPath().toString();
String line = value.toString();
if (line == null || line.equals("")) return;
if (filepath.indexOf("employee") != -1) {
String[] lines = line.split(DELIMITER);
if(lines.length < 2) return;
String company_id = lines[0];
String employee = lines[1];
context.write(new Text(company_id),new Text("a:"+employee));
}
else if(filepath.indexOf("salary") != -1) {
String[] lines = line.split(DELIMITER);
if(lines.length < 2) return;
String company_id = lines[0];
String salary = lines[1];
context.write(new Text(company_id), new Text("b:" + salary));
}
}
}
public static class LeftJoinReduce extends
Reducer<Text, Text, Text, Text> {
protected void reduce(Text key, Iterable<Text> values,
Context context) throws IOException, InterruptedException{
Vector<String> vecA = new Vector<String>();
Vector<String> vecB = new Vector<String>();
for(Text each_val:values) {
String each = each_val.toString();
if(each.startsWith("a:")) {
vecA.add(each.substring(2));
} else if(each.startsWith("b:")) {
vecB.add(each.substring(2));
}
}
for (int i = 0; i < vecA.size(); i++) {
/*
* 如果vecB為空的話，将A裡的輸出，B的位置補null。
*/
if (vecB.size() == 0) {
context.write(key, new Text(vecA.get(i) + DELIMITER + "null"));
} else {
for (int j = 0; j < vecB.size(); j++) {
context.write(key, new Text(vecA.get(i) + DELIMITER + vecB.get(j)));
}
}
}
}
}
public int run(String[] args) throws Exception {
Configuration conf = getConf();
GenericOptionsParser optionparser = new GenericOptionsParser(conf, args);
conf = optionparser.getConfiguration();
Job job = new Job(conf,"leftjoin");
job.setJarByClass(LeftJoin.class);
FileInputFormat.addInputPaths(job, conf.get("input_dir"));
Path out = new Path(conf.get("output_dir"));
FileOutputFormat.setOutputPath(job, out);
job.setNumReduceTasks(conf.getInt("reduce_num",2));
job.setMapperClass(LeftJoinMapper.class);
job.setReducerClass(LeftJoinReduce.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
conf.set("mapred.textoutputformat.separator", ",");
return (job.waitForCompletion(true) ? 0 : 1);
}
public static void main(String[] args) throws Exception{
int res = ToolRunner.run(new Configuration(),new LeftJoin(),args);
System.exit(res);
}
}
稍微說一下處理邏輯：
1.map階段，把所有輸入拆分為k,v形式。其中k是company_id，即我們要關聯的字段。如果輸入是employee相關的檔案，那麼map階段的value加上辨別符"a",表示是employee的輸出。對于salary檔案，加上辨別符"b"。
2.reduce階段，将每個k下的value清單拆分為分别來自employee和salary的兩部分，然後雙層循環做笛卡爾積即可。
3.注意的是，因為是left join，是以在reduce階段，如果employee對應的company_id有，而salary沒有，注意要輸出此部分資料。
3.打包
将上面的項目用maven打包，并上傳到伺服器上。
4.使用shell腳本run起來
在伺服器上寫一個最簡單的shell腳本，将代碼run起來：
vim run_join.sh
#!/bin/bash
output=/tmp/wanglei/leftjoin
if hadoop fs -test -d $output
then
hadoop fs -rm -r $output
fi
hadoop jar testjoin.jar leilei.bit.edu.testjoin.LeftJoin \
-Dinput_dir=/tmp/wanglei/employee,/tmp/wanglei/salary \
-Doutput_dir=$output \
-Dmapred.textoutputformat.separator=","
執行此shell腳本：
./run_join.sh
5.最終輸出
等job跑完以後，檢視最終的輸出結果：
hadoop fs -cat /tmp/wanglei/leftjoin/*
elong,ali,2000
elong,xiaoming,2000
tengxun,lilei,2200
tengxun,xiaoming,2200
jd,mike,1600
jd,david,1600
tb,lucifer,1800
tb,mike,1800
xxx,aaa,null
最終的輸出結果，準确無誤！至此，我們用mr完美實作了left join的功能！
![Windows下Cygwin環境的Hadoop安裝（3）- 運作hadoop中的wordcount執行個體遇到的問題和解決方法[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)