MapReduce实现join操作

在MapReduce中实现join操作通常有两种方式：Reduce-side join和Map-side join。

1. Reduce-side join：

Reduce-side join是最常用的实现方式。它的基本思想是将两个需要join的表分别映射为(key, value)的形式，其中key为需要join的字段，value则包含该字段以及其他需要输出的字段。然后将两个表的数据都输入到Map函数中，在Map函数中对两个表的数据进行标记，并将需要join的字段作为输出的key。在Reduce函数中，对相同的key进行合并，得到最终的输出结果。

下面是一个示例的Reduce-side join实现：

Map函数：

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String line = value.toString();
    String[] fields = line.split(",");
    String joinKey = fields[0];
    String table = fields[1]; // 表名
    String data = fields[2]; // 数据
    Text outputKey = new Text(joinKey);
    Text outputValue = new Text(table + ":" + data);
    context.write(outputKey, outputValue);
}

Reduce函数：

public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    List<String> table1Data = new ArrayList<String>();
    List<String> table2Data = new ArrayList<String>();
    for (Text value : values) {
        String[] fields = value.toString().split(":");
        if (fields[0].equals("table1")) {
            table1Data.add(fields[1]);
        } else if (fields[0].equals("table2")) {
            table2Data.add(fields[1]);
        }
    }
    for (String data1 : table1Data) {
        for (String data2 : table2Data) {
            context.write(key, new Text(data1 + "," + data2));
        }
    }
}

2. Map-side join：

Map-side join是一种更加高效的实现方式，它的基本思想是将一个表的数据缓存到内存中，然后在Map函数中将另一个表的数据与缓存的数据进行join。需要注意的是，Map-side join只适用于小表与大表之间的join操作，因为需要将小表的数据全部缓存到内存中。

下面是一个示例的Map-side join实现：

Map函数：

public void setup(Context context) throws IOException, InterruptedException {
    // 读取小表的数据并缓存到内存中
    BufferedReader br = new BufferedReader(new FileReader("table1.csv"));
    String line;
    while ((line = br.readLine()) != null) {
        String[] fields = line.split(",");
        String joinKey = fields[0];
        String data = fields[1] + "," + fields[2];
        table1Data.put(joinKey, data);
    }
    br.close();
}

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String line = value.toString();
    String[] fields = line.split(",");
    String joinKey = fields[0];
    String data = fields[1] + "," + fields[2];
    if (table1Data.containsKey(joinKey)) {
        String table1Data = table1Data.get(joinKey);
        context.write(new Text(joinKey), new Text(table1Data + "," + data));
    }
}

需要注意的是，Map-side join需要提前将小表的数据缓存到内存中，因此需要在Map函数之前执行setup函数。同时，为了提高效率，通常使用HashMap等数据结构来缓存小表的数据。