AI 助理
备案控制台
开发者社区 大数据 文章 正文

Hadoop-11-MapReduce JOIN 操作的Java实现 Driver Mapper Reducer具体实现逻辑 模拟SQL进行联表操作

2024-10-13 42
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： Hadoop-11-MapReduce JOIN 操作的Java实现 Driver Mapper Reducer具体实现逻辑 模拟SQL进行联表操作

章节内容

上一节我们完成了：


MapReduce的介绍

Hadoop序列化介绍

Mapper编写规范

Reducer编写规范

Driver编写规范

WordCount功能开发

WordCount本地测试

背景介绍

这里是三台公网云服务器，每台 2C4G，搭建一个Hadoop的学习环境，供我学习。

之前已经在 VM 虚拟机上搭建过一次，但是没留下笔记，这次趁着前几天薅羊毛的3台机器，赶紧尝试在公网上搭建体验一下。


注意，如果你和我一样，打算用公网部署，那一定要做好防火墙策略，避免不必要的麻烦！！！

请大家都以学习为目的，也请不要对我的服务进行嗅探或者攻击！！！


但是有一台公网服务器我还运行着别的服务，比如前几天发的：autodl-keeper 自己写的小工具，防止AutoDL机器过期的。还跑着别的Web服务，所以只能挤出一台 2C2G 的机器。那我的配置如下了：


2C4G 编号 h121

2C4G 编号 h122

2C2G 编号 h123

业务需求

平常我们在业务上，有很多时候表都是分开的，通过一些 id 或者 code 来进行关联。

在大数据的情况下，也有很多这种情况，我们需要进行联表操作。

表1

项目编码projectCode 项目名projectName
• 1

表2

项目编码projectCode 项目类型projectType 项目分类projectFrom

在 SQL 中，可以通过 LEFT JOIN 来实现字段补齐。大数据下，也需要进行这样的操作，我们需要借助 MapReduce。

表1测试

"8aea9ba2-435c-48bd-9751-1cbd4c344d4e"  "社区项目1"
"02d9c090-e467-42b6-9c14-52cacd72a4a8"  "社区项目2"
"244dcaca-0778-4eec-b3a2-403f8fac1dfb"  "智慧社区"
"94befb97-d1af-43f2-b5d5-6df9ce5b9393"  "公交站点"
"f44c8d10-bc92-4398-ad9b-8c11dd48ad7c"  "街道布建"
"2e556d83-bb56-45b1-8d6e-00510902c464"  "街道公交站点"
"3ba00542-eac9-4399-9c2b-3b06e671f4c9"  "未命名项目1"
"5a5982d7-7257-422f-822a-a0c2f31c28d1"  "未命名项目2"

表2测试

"8aea9ba2-435c-48bd-9751-1cbd4c344d4e"  "重要类型"  "种类1"
"02d9c090-e467-42b6-9c14-52cacd72a4a8"  "重要类型"  "种类1"
"244dcaca-0778-4eec-b3a2-403f8fac1dfb"  "重要类型"  "种类1"
"94befb97-d1af-43f2-b5d5-6df9ce5b9393"  "普通类型"  "种类1"
"f44c8d10-bc92-4398-ad9b-8c11dd48ad7c"  "普通类型"  "种类2"
"2e556d83-bb56-45b1-8d6e-00510902c464"  "普通类型"  "种类2"
"3ba00542-eac9-4399-9c2b-3b06e671f4c9"  "一般类型"  "种类2"
"5a5982d7-7257-422f-822a-a0c2f31c28d1"  "一般类型"  "种类2"

SQL连表

假设我们使用SQL的方式联表：

SELECT
  *
FROM
  t_project
LEFT JOIN
  t_project_info
ON
  t_project.projectCode=t_project_info.projectCode

Reduce JOIN

有时候，表可能过大，无法支持我们使用 SQL 进行连表查询。

这里我们编写一个程序来完成操作。

ProjectBean

这里是最终的Bean类，里边是两个表把字段补齐的结果，一会儿我们将使用这个类进行表的连接。

package icu.wzk.demo03;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class ProjectBean implements Writable {

    private String projectCode;

    private String projectName;

    private String projectType;

    private String projectFrom;

    private String flag;

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(projectCode);
        dataOutput.writeUTF(projectName);
        dataOutput.writeUTF(projectType);
        dataOutput.writeUTF(projectFrom);
        dataOutput.writeUTF(flag);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.projectCode = dataInput.readUTF();
        this.projectName = dataInput.readUTF();
        this.projectType = dataInput.readUTF();
        this.projectFrom = dataInput.readUTF();
        this.flag = dataInput.readUTF();
    }

    public ProjectBean(String projectCode, String projectName, String projectType, String projectFrom, String flag) {
        this.projectCode = projectCode;
        this.projectName = projectName;
        this.projectType = projectType;
        this.projectFrom = projectFrom;
        this.flag = flag;
    }

    public ProjectBean() {

    }

    @Override
    public String toString() {
        return "ProjectBean{" +
                "projectCode='" + projectCode + '\'' +
                ", projectName='" + projectName + '\'' +
                ", projectType='" + projectType + '\'' +
                ", projectFrom='" + projectFrom + '\'' +
                ", flag=" + flag + '\'' +
                '}';
    }

    public String getProjectCode() {
        return projectCode;
    }

    public void setProjectCode(String projectCode) {
        this.projectCode = projectCode;
    }

    public String getProjectName() {
        return projectName;
    }

    public void setProjectName(String projectName) {
        this.projectName = projectName;
    }

    public String getProjectType() {
        return projectType;
    }

    public void setProjectType(String projectType) {
        this.projectType = projectType;
    }

    public String getProjectFrom() {
        return projectFrom;
    }

    public void setProjectFrom(String projectFrom) {
        this.projectFrom = projectFrom;
    }

    public String getFlag() {
        return flag;
    }

    public void setFlag(String flag) {
        this.flag = flag;
    }
}

Reduce Driver

package icu.wzk.demo03;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class ReducerJoinDriver {

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

        // String inputPath = args[0];
        // String outputPath = args[1];

        // === 测试环境 ===
        String inputPath = "project_test";
        String outputPath = "project_test_output";
        // === ===

        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration, "ReducerJoinDriver");
        job.setJarByClass(ReducerJoinDriver.class);

        job.setMapperClass(ReducerJoinMapper.class);
        job.setReducerClass(ReducerJoinReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(ProjectBean.class);

        job.setOutputKeyClass(ProjectBean.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }

}

ReduceMapper

package icu.wzk.demo03;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class ReducerJoinMapper extends Mapper<LongWritable, Text, Text, ProjectBean> {

    String name;
    ProjectBean projectBean = new ProjectBean();
    Text k = new Text();

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, ProjectBean>.Context context) throws IOException, InterruptedException {
        // 获取路径信息
        name = context.getInputSplit().toString();
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, ProjectBean>.Context context) throws IOException, InterruptedException {
        String line = value.toString();
        if (name.contains("layout_project")) {
            // layout_project
            String[] fields = line.split("\t");
            projectBean.setProjectCode(fields[0]);
            projectBean.setProjectName(fields[1]);
            projectBean.setProjectType("");
            projectBean.setProjectFrom("");
            projectBean.setFlag("layout_project");
            // projectCode 关联
            k.set(fields[0]);
        } else {
            // project_info
            String[] fields = line.split("\t");
            projectBean.setProjectCode(fields[0]);
            projectBean.setProjectName("");
            projectBean.setProjectType(fields[1]);
            projectBean.setProjectFrom(fields[2]);
            projectBean.setFlag("project_info");
            // projectCode 关联
            k.set(fields[0]);
        }
        context.write(k, projectBean);
    }
}

ReduceReducer

package icu.wzk.demo03;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class ReducerJoinReducer extends Reducer<Text, ProjectBean, ProjectBean, NullWritable> {

    @Override
    protected void reduce(Text key, Iterable<ProjectBean> values, Reducer<Text, ProjectBean, ProjectBean, NullWritable>.Context context) throws IOException, InterruptedException {
        List<ProjectBean> dataList = new ArrayList<>();
        ProjectBean deviceProjectBean = new ProjectBean();
        for (ProjectBean pb : values) {
            if ("layout_project".equals(pb.getFlag())) {
                // layout_project
                ProjectBean projectProjectBean = new ProjectBean(
                        pb.getProjectCode(),
                        pb.getProjectName(),
                        pb.getProjectType(),
                        pb.getProjectFrom(),
                        pb.getFlag()
                );
                dataList.add(projectProjectBean);
            } else {
                // project_info
                deviceProjectBean = new ProjectBean(
                        pb.getProjectCode(),
                        pb.getProjectName(),
                        pb.getProjectType(),
                        pb.getProjectFrom(),
                        pb.getFlag()
                );
            }
        }

        for (ProjectBean pb : dataList) {
            pb.setProjectType(deviceProjectBean.getProjectType());
            pb.setProjectFrom(deviceProjectBean.getProjectFrom());
            context.write(pb, NullWritable.get());
        }
    }
}

运行结果

ProjectBean{projectCode='"02d9c090-e467-42b6-9c14-52cacd72a4a8"', projectName='"社区项目2"', projectType='"重要类型"', projectFrom='"种类1"', flag=layout_project'}
ProjectBean{projectCode='"244dcaca-0778-4eec-b3a2-403f8fac1dfb"', projectName='"智慧社区"', projectType='"重要类型"', projectFrom='"种类1"', flag=layout_project'}
ProjectBean{projectCode='"2e556d83-bb56-45b1-8d6e-00510902c464"', projectName='"街道公交站点"', projectType='"普通类型"', projectFrom='"种类2"', flag=layout_project'}
ProjectBean{projectCode='"3ba00542-eac9-4399-9c2b-3b06e671f4c9"', projectName='"未命名项目1"', projectType='"一般类型"', projectFrom='"种类2"', flag=layout_project'}
ProjectBean{projectCode='"5a5982d7-7257-422f-822a-a0c2f31c28d1"', projectName='"未命名项目2"', projectType='"一般类型"', projectFrom='"种类2"', flag=layout_project'}
ProjectBean{projectCode='"8aea9ba2-435c-48bd-9751-1cbd4c344d4e"', projectName='"社区项目1"', projectType='"重要类型"', projectFrom='"种类1"', flag=layout_project'}
ProjectBean{projectCode='"94befb97-d1af-43f2-b5d5-6df9ce5b9393"', projectName='"公交站点"', projectType='"普通类型"', projectFrom='"种类1"', flag=layout_project'}
ProjectBean{projectCode='"f44c8d10-bc92-4398-ad9b-8c11dd48ad7c"', projectName='"街道布建"', projectType='"普通类型"', projectFrom='"种类2"', flag=layout_project'}

方案缺点

JOIN 操作是在 reduce 阶段完成的，reduce端处理压力过大，map节点的运算负载很低，资源利用率不高。

文章标签：
SQL
Java
分布式计算
Hadoop
大数据
关键词：
Java SQL
hadoop java
mapper SQL
hadoop sql
hadoop实现逻辑
武子康
目录
相关文章
刘大猫.
|
3月前
|
分布式计算 Java Hadoop
java使用hbase、hadoop报错举例
java使用hbase、hadoop报错举例
刘大猫.
120 4
武子康
|
2月前
|
分布式计算 资源调度 Hadoop
Hadoop-10-HDFS集群 Java实现MapReduce WordCount计算 Hadoop序列化 编写Mapper和Reducer和Driver 附带POM 详细代码 图文等内容
Hadoop-10-HDFS集群 Java实现MapReduce WordCount计算 Hadoop序列化 编写Mapper和Reducer和Driver 附带POM 详细代码 图文等内容
武子康
104 3
武子康
|
2月前
|
分布式计算 资源调度 Hadoop
大数据-01-基础环境搭建 超详细 Hadoop Java 环境变量 3节点云服务器 2C4G XML 集群配置 HDFS Yarn MapRedece
大数据-01-基础环境搭建 超详细 Hadoop Java 环境变量 3节点云服务器 2C4G XML 集群配置 HDFS Yarn MapRedece
武子康
88 4
武子康
|
2月前
|
分布式计算 NoSQL Java
Hadoop-32 ZooKeeper 分布式锁问题 分布式锁Java实现 附带案例和实现思路代码
Hadoop-32 ZooKeeper 分布式锁问题 分布式锁Java实现 附带案例和实现思路代码
武子康
51 2
武子康
|
2月前
|
分布式计算 Java Hadoop
Hadoop-30 ZooKeeper集群 JavaAPI 客户端 POM Java操作ZK 监听节点 监听数据变化 创建节点 删除节点
Hadoop-30 ZooKeeper集群 JavaAPI 客户端 POM Java操作ZK 监听节点 监听数据变化 创建节点 删除节点
武子康
69 1
小王老师呀
|
3月前
|
分布式计算 数据库
Mapreduce中的Mapper&reducer
【9月更文挑战第19天】在 MapReduce 框架中，Mapper 和 Reducer 是处理大规模数据集的关键组件。Mapper 负责将输入数据分割成键值对，而 Reducer 则对这些键值对进行汇总处理，生成最终结果。两者通过并行处理和分布式计算协同工作，Mapper 将数据转换为键值对，Reducer 对相同键的值进行聚合。开发人员需实现相应接口并编写定制逻辑，以充分利用框架优势，处理大规模数据集并获得有价值的结果。
小王老师呀
129 7
技术小胖子
|
Java
Java多线程初学者指南（5）：join方法的使用
技术小胖子
877 0
geekori
|
Java
Java多线程初学者指南（5）：join方法的使用
本文为原创，如需转载，请注明作者和出处，谢谢！     在上面的例子中多次使用到了Thread类的join方法。我想大家可能已经猜出来join方法的功能是什么了。
geekori
785 0
蓝易云
|
5天前
|
安全 Java API
java如何请求接口然后终止某个线程
通过本文的介绍，您应该能够理解如何在Java中请求接口并根据返回结果终止某个线程。合理使用标志位或 `interrupt`方法可以确保线程的安全终止，而处理好网络请求中的各种异常情况，可以提高程序的稳定性和可靠性。
蓝易云
35 6
游客5fdji2pvmf8888
|
20天前
|
设计模式 Java 开发者
Java多线程编程的陷阱与解决方案####
本文深入探讨了Java多线程编程中常见的问题及其解决策略。通过分析竞态条件、死锁、活锁等典型场景，并结合代码示例和实用技巧，帮助开发者有效避免这些陷阱，提升并发程序的稳定性和性能。 ####
游客5fdji2pvmf8888
52 16

热门文章

最新文章

  • 1
    【Java多数据源实现教程】实现动态数据源、多数据源切换方式（上）
  • 2
    Java-学习笔记-8-接口
  • 3
    java中的工厂模式
  • 4
    超简单的java爬虫
  • 5
    Java 学习路线 2024 最新版！
  • 6
    从零搭建Web所需服务（四）Java连接ES进行分词并获取结果
  • 7
    迅速了解多例模式竞争者-枚举 | 带你学《Java面向对象编程》之七十四
  • 8
    带你读《Java程序设计与计算思维》之三：流程控制
  • 9
    java源码 - ReentrantLock
  • 10
    Java Poi 创建与读取Excel
  • 1
    SQL脚本相除
    72
  • 2
    0基础学习SQL注入之万能账号密码（BUUctf例题-[极客大挑战 2019]EasySQL1）
    669
  • 3
    SQL脚本利用带关联子查询Update语句更新数据
    135
  • 4
    10分钟了解Flink SQL使用
    392
  • 5
    SQL脚本把多行SQL数据变成一条多列数据
    120
  • 6
    mybatis动态sql
    44
  • 7
    基于若依的ruoyi-nbcio流程管理系统修改代码生成的sql菜单id修改成递增id（谨慎修改，大并发分布式有弊端）
    114
  • 8
    SQL脚本字符串替换
    77
  • 9
    你写的每条SQL都是全表扫描吗
    186
  • 10
    数据库SQL语言实战（六）
    54

    • 相关课程

    更多
  • Java面试疑难点解析 - 面试技巧及语言基础
  • Java面试疑难点解析 - Java Web开发
  • Java面试疑难点解析 - 系统架构及项目设计
  • Java编程入门
  • Java面向对象编程
  • Java高级编程

    • 相关电子书

    更多
  • Spring Cloud Alibaba - 重新定义 Java Cloud-Native
  • The Reactive Cloud Native Arch
  • JAVA开发手册1.5.0

    • 相关实验场景

    更多
  • 一小时快速掌握 SQL 语法
  • PolarDB for AI：在数据库中通过SQL实现AI能力
  • 玩转MaxCompute SQL! 30分钟搞定数据分析挖掘
  • 阿里云平台上进行Java程序的编译与运行
  • 使用SQL语句实现数据查询操作
  • 使用SQL语句管理索引
    • 下一篇
    阿里云oss存储简介和如何使用