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

对给定的数据利用MapReduce编程实现数据的清洗和预处理，编程实现数据存储到HBase数据库，实现数据的增删改查操作接口

2023-12-26 203
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云原生网关 MSE Higress，422元/月
服务治理 MSE Sentinel/OpenSergo，Agent数量 不受限
MSE Nacos/ZooKeeper 企业版试用，1600元额度，限量50份
简介： 对给定的数据利用MapReduce编程实现数据的清洗和预处理，编程实现数据存储到HBase数据库，实现数据的增删改查操作接口


对给定的数据利用MapReduce编程实现数据的清洗和预处理，编程实现数据存储到HBase数据库，实现数据的增删改查操作接口，同时对MapReduce处理好的数据利用Hive实现数据的基本统计。

设计要求：

  1. 根据数据特征，设计一个任务场景，利用MapReduce编程实现数据的清洗和预处理。（10分）
  2. 利用HDFS的JavaAPI编写程序将原始数据和预处理后的数据上传到分布式文件系统

数据集:

链接：https://pan.baidu.com/s/1rnUJn5ld45HpLhzbwYIM1A

提取码：7bsd

package com.company.HDFS;
import java.io.IOException;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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 step0 {
  final static String INPUT_PATH="hdfs://192.168.88.100/data";
  final static String OUTPUT_PATH="hdfs://192.168.88.100/output";
  public static void main(String[] args) throws Exception {
    // TODO Auto-generated method stub
    Configuration configuration = new Configuration();
    FileSystem fileSystem =FileSystem.get(new URI(INPUT_PATH),configuration);
    if (fileSystem.exists(new Path(OUTPUT_PATH))) {
      fileSystem.delete(new Path(OUTPUT_PATH),true);
    }
    Job job = new Job(configuration,"step0");
    FileInputFormat.setInputPaths(job, INPUT_PATH);
    FileOutputFormat.setOutputPath(job,new Path(OUTPUT_PATH));
    job.setJarByClass(step0.class);
    job.setMapperClass(ReMapper.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(NullWritable.class);
    job.setReducerClass(ReReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(NullWritable.class);
    job.waitForCompletion(true);
  }
    public static class ReReducer extends Reducer<Text,NullWritable, Text,NullWritable> {
        private IntWritable result = new IntWritable();
        public ReReducer() {
        }
        protected void reduce(Text key2, Iterable<NullWritable> value2, Reducer<Text,NullWritable, Text,NullWritable>.Context context) throws IOException, InterruptedException {
            context.write(key2,NullWritable.get());
        }
    }
    public static class ReMapper extends Mapper<LongWritable, Text, Text,NullWritable> {
        private static final int FAIL_DATA=9999;
        public void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
          context.write(value,NullWritable.get());
        }
    }
}
• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
package com.company.HDFS;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
public class step1 {
    /**
     * 查看   所有文件
     */
    @Test
    public void demo_03() {
        try {
            //1 获取文件系统
            Configuration configuration = new Configuration();
            FileSystem fs = FileSystem.get(new URI("hdfs://192.168.88.100:8020"), configuration, "root");
            // 2 获取文件详情
            RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
            while (listFiles.hasNext()) {
                LocatedFileStatus status = listFiles.next();
                // 输出详情
                // 文件名称
                System.out.println(status.getPath().getName());
                // 长度
                System.out.println(status.getLen());
                // 权限
                System.out.println(status.getPermission());
                // 分组
                System.out.println(status.getGroup());
                // 获取存储的块信息
                BlockLocation[] blockLocations = status.getBlockLocations();
                for (BlockLocation blockLocation : blockLocations) {
                    // 获取块存储的主机节点
                    String[] hosts = blockLocation.getHosts();
                    for (String host : hosts) {
                        System.out.println(host);
                    }
                }
                System.out.println("-----------分割线----------");
            }
            // 3 关闭资源
            fs.close();
        } catch (Exception ex) {
        }
    }
    /**
     * 上传
     */
    @Test
    public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        configuration.set("dfs.replication", "2");
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.88.100:8020"), configuration, "root");
        // 2 上传文件
        fs.copyFromLocalFile(new Path("J:\\the_efforts_paid_offf\\HDFS_HBase_HiveApi\\src\\main\\java\\com\\company\\datas\\iris.data"), new Path("hdfs://192.168.88.100/input"));
        // 3 关闭资源
        fs.close();
        System.out.println("over");
    }
}
• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79
• 80
• 81
• 82
• 83
• 84
• 85
package com.company.HDFS;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Test;
import org.testng.annotations.AfterTest;
import org.testng.annotations.BeforeTest;
import java.io.IOException;
/**
 * @author ChinaManor
 * #Description hbase的javaAPI
 * #Date: 2021/12/19 18:10
 */
public class step2 {
    /**
     * @Description: createTable():创建表的方法
     * @Param: 0
     * @return: 0
     */
    @Test
    public void createTable() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node1:2181");
        //建立连接
        Connection conn = ConnectionFactory.createConnection(conf);
        //获取表的管理类
        Admin admin = conn.getAdmin();
        //定义表
        HTableDescriptor hTableDescriptor=new HTableDescriptor(TableName.valueOf("demo"));
        //定义列簇
        HColumnDescriptor hColumnDescriptor =new HColumnDescriptor("info");
        //讲列簇定义到表中
        hTableDescriptor.addFamily(hColumnDescriptor);
        //执行建表操作
        admin.createTable(hTableDescriptor);
        admin.close();
        conn.close();
    }
    /**
     * @Description: 向Hbase中插入数据的方法
     * @Param: null
     * @return: null
     */
    @Test
    public void put(){
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","node1:2181");
        try {
            //建立连接
            Connection conn= ConnectionFactory.createConnection(conf);
            //获取表
            Table table=conn.getTable(TableName.valueOf("demo"));
            //用行键实例化put
            Put put= new Put("rk001".getBytes());
            //指定列簇名，列名，和值
            put.addColumn("info".getBytes(),"name".getBytes(),"zhangsan".getBytes());
            table.put(put);
            table.close();
            conn.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * @Description: scan()查询一个表的所有信息
     * @Param: 1
     * @return: 1
     */
    @Test
    public  void scan() throws IOException {
        Configuration conf=HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node1:2181");
        //建立连接
        Connection conn=ConnectionFactory.createConnection(conf);
        //获取表
        Table table=conn.getTable(TableName.valueOf("demo"));
        //初始化Scan实例
        Scan scan=new Scan();
        //增加过滤条件
        scan.addColumn("info".getBytes(), "name".getBytes());
        //返回结果
        ResultScanner rss=table.getScanner(scan);
        //迭代并取出结果
        for(Result rs:rss){
            String valStr=Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            System.out.println(valStr);
        }
        //关闭连接
        table.close();
        conn.close();
    }
    /**
     * @Description: delete()删除表中的信息
     * @Param: 1
     * @return: 1
     */
    @Test
    public  void delete() throws IOException {
        Configuration conf=HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node1:2181");
        //建立连接
        Connection conn=ConnectionFactory.createConnection(conf);
        //获取表
        Table table=conn.getTable(TableName.valueOf("demo"));
        // 用行键来实例化Delete实例
        Delete del = new Delete("rk0001".getBytes());
        // 执行删除
        table.delete(del);
        //关闭连接
        table.close();
        conn.close();
    }
}
• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79
• 80
• 81
• 82
• 83
• 84
• 85
• 86
• 87
• 88
• 89
• 90
• 91
• 92
• 93
• 94
• 95
• 96
• 97
• 98
• 99
• 100
• 101
• 102
• 103
• 104
• 105
• 106
• 107
• 108
• 109
• 110
• 111
• 112
• 113
• 114
• 115
• 116
• 117
• 118
• 119
• 120
• 121
• 122
• 123
• 124
• 125
• 126
• 127
• 128
• 129
• 130
• 131
• 132
• 133
• 134
• 135
• 136
• 137
• 138
• 139
• 140
• 141
• 142
• 143
• 144


文章标签：
云数据库HBase版
微服务引擎
分布式计算
Hbase
数据库
存储
分布式数据库
关键词：
数据库数据
数据库增删改查
云数据库HBase版数据库
hbase数据库
数据库数据存储
Maynor
目录
相关文章
蓝易云
|
6月前
|
SQL 数据库 开发者
Python中使用Flask-SQLAlchemy对数据库的增删改查简明示例
这样我们就对Flask-SQLAlchemy进行了一次简明扼要的旅程，阐述了如何定义模型，如何创建表，以及如何进行基本的数据库操作。希望你在阅读后能对Flask-SQLAlchemy有更深入的理解，这将为你在Python世界中从事数据库相关工作提供极大的便利。
蓝易云
650 77
计蒙不吃鱼
|
6月前
|
XML 数据库 Android开发
Android数据库的使用（增删改查）
本文介绍了一个简单的数据库操作Demo，包含创建数据库、增删改查功能。通过5个按钮分别实现创建数据库、插入数据、删除数据、更新数据和查询数据的操作。代码结构清晰，适合初学者学习Android SQLite数据库基础操作。
计蒙不吃鱼
203 5
计蒙不吃鱼
|
6月前
|
数据库 Android开发 开发者
Android常用的room增删改查语句（外部数据库）
本文分享了将一个原生数据库驱动的单词APP重构为使用Room库的过程及遇到的问题，重点解决了Room中增删改查的常用语句实现。文章通过具体示例（以“forget”表为例），详细展示了如何定义实体类、Dao接口、Database类以及Repository和ViewModel的设计与实现。同时，提供了插入、删除、更新和查询数据的代码示例，包括模糊查询、分页加载等功能。此外，针对外部数据库导入问题，作者建议可通过公众号“计蒙不吃鱼”获取更多支持。此内容适合有一定Room基础的开发者深入学习。
计蒙不吃鱼
226 0
Android常用的room增删改查语句（外部数据库）
蓝易云
|
8月前
|
SQL 数据库连接 Linux
数据库编程：在PHP环境下使用SQL Server的方法。
看看你吧，就像一个调皮的小丑鱼在一片广阔的数据库海洋中游弋，一路上吞下大小数据如同海中的珍珠。不管有多少难关，只要记住这个流程，剩下的就只是探索未知的乐趣，沉浸在这个充满挑战的数据库海洋中。
蓝易云
207 16
蓝易云
|
9月前
|
中间件 关系型数据库 数据库
docker快速部署OS web中间件 数据库 编程应用
通过Docker，可以轻松地部署操作系统、Web中间件、数据库和编程应用。本文详细介绍了使用Docker部署这些组件的基本步骤和命令，展示了如何通过Docker Compose编排多容器应用。希望本文能帮助开发者更高效地使用Docker进行应用部署和管理。
蓝易云
278 19
心疼你的一切
|
9月前
|
关系型数据库 MySQL 数据库连接
Unity连接Mysql数据库 增 删 改 查
在 Unity 中连接 MySQL 数据库，需使用 MySQL Connector/NET 作为数据库连接驱动，通过提供服务器地址、端口、用户名和密码等信息建立 TCP/IP 连接。代码示例展示了如何创建连接对象并执行增删改查操作，确保数据交互的实现。测试代码中，通过 `MySqlConnection` 类连接数据库，并使用 `MySqlCommand` 执行 SQL 语句，实现数据的查询、插入、删除和更新功能。
心疼你的一切
526 2
程序员一鸣
|
11月前
|
SQL 关系型数据库 API
HarmonyOs开发：关系型数据库封装之增删改查
每个方法都预留了多种调用方式，比如使用callback异步回调或者使用Promise异步回调，亦或者同步执行，大家在使用的过程中，可以根据自身业务需要进行选择性调用，也分别暴露了成功和失败的方法，可以针对性的判断在执行的过程中是否执行成功。
程序员一鸣
405 13
奔跑的数据
|
11月前
|
存储 数据采集 数据挖掘
CSV vs 数据库：数据存储的最佳选择是什么
本文介绍了爬虫数据存储中CSV和数据库的优缺点，分析了两者在不同场景下的适用性。CSV简单易用、资源消耗低，适合小量数据；数据库则在处理大量数据和复杂查询时表现出色，支持并发操作。通过Python代码示例，展示了如何使用多线程和爬虫代理IP技术将百度搜索数据存储到MySQL数据库中，适用于大型项目和复杂数据分析需求。
奔跑的数据
709 2
三三是该溜子
|
11月前
|
SQL Java 数据库连接
JDBC编程安装———通过代码操控数据库
本文，教你从0开始学习JBCD，包括驱动包的下载安装调试设置，以及java是如何通过JBDC实现对数据库的操作，以及代码的分析，超级详细
三三是该溜子
168 0
众所周知
|
数据库连接 Go 数据库
Go语言中的错误注入与防御编程。错误注入通过模拟网络故障、数据库错误等，测试系统稳定性
本文探讨了Go语言中的错误注入与防御编程。错误注入通过模拟网络故障、数据库错误等，测试系统稳定性；防御编程则强调在编码时考虑各种错误情况，确保程序健壮性。文章详细介绍了这两种技术在Go语言中的实现方法及其重要性，旨在提升软件质量和可靠性。
众所周知
189 1

热门文章

最新文章

  • 1
    除了JDBC，还有哪些常见的数据库访问技术？
  • 2
    云栖大会｜AI时代关系型数据库创新突破：软硬协同赋能企业数智化转型
  • 3
    云栖大会｜AI时代的数据库变革升级与实践：Data+AI驱动企业智能新范式
  • 4
    阿里云RDS云数据库全解析：产品功能、收费标准与活动参考
  • 5
    AI大模型应用实践 八：如何通过RAG数据库实现大模型的私有化定制与优化
  • 6
    云栖重磅｜瑶池数据库：从云原生数据底座向“AI就绪”的多模态数据底座演进
  • 7
    通义灵码产品演示： 数据库设计与数据分析
  • 8
    云栖大会｜AI浪潮下的NoSQL演进：下一代数据库的破局之道
  • 9
    Python SQLAlchemy模块：从入门到实战的数据库操作指南
  • 10
    免费体验！高效实现自建 MySQL 数据库平滑迁移至 PolarDB-X
  • 1
    【赵渝强老师】HBase的物理存储结构
    429
  • 2
    【赵渝强老师】HBase的逻辑存储结构
    236
  • 3
    Ubuntu22.04下搭建Hadoop3.3.6+Hbase2.5.6+Phoenix5.1.3开发环境的指南
    434
  • 4
    【赵渝强老师】HBase的体系架构
    374
  • 5
    分布式存储数据恢复—hbase和hive数据库数据恢复案例
    271
  • 6
    在 HBase 集群中，Prometheus 通常监控哪些类型的性能指标？
    682
  • 7
    如何使用 HBase Shell 进行数据的实时监控和备份？
    266
  • 8
    如何使用 HBase Shell 进行数据的批量导入和导出？
    883
  • 9
    【赵渝强老师】HBase的体系架构
    584
  • 10
    【赵渝强老师】HBase的表结构
    474

    • 相关课程

    更多
  • 数据库的前世今生
  • 数据库核心概念
  • 从传统数据库到云数据库演进
  • 数据库常见问题排查
  • 数据库及SQL/MySQL基础
  • 高校精品课-西安交通大学 -数据库理论与技术

    • 相关电子书

    更多
  • 大数据时代的存储 ——HBase的实践与探索
  • Hbase在滴滴出行的应用场景和最佳实践
  • 阿里云HBase主备双活
    • 下一篇
    拔俗AI临床大数据科研分析平台：让医学研究更智能、更高效