功能说明.md · itzlg/phase-four-module-five

#### 1.MapReduce程序计算多个文件，对里面的数字进行排序，并输出

需求：将file1.txt、file2.txt、file3.txt中的记录整合排序后，输出到一个文件中，包含行号。

实现：编写MapReduce程序，实现上述内容。

分析：利用MR的sort能力，必须进行shuffle，一定实现reduce；

- 编写mapper：将<k1,v1>(行偏移量，行值) --> <k2,v2> (行值，1)
- 编写reducer：接收来自mapper端的数据<k2,[v2,v2,v2...]>,此时，数据已经按key进行排序。循环遍历values，context.write(linesum,k2);linesum++;
- 编写dirver

代码实现：

依赖：

```xml
<dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-core</artifactId>
    <version>2.8.2</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>2.9.2</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>2.9.2</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-hdfs</artifactId>
    <version>2.9.2</version>
</dependency>
```

NumberMapper类：

```java
public class NumberMapper extends Mapper<Object, Text, IntWritable, IntWritable> {

private static IntWritable data = new IntWritable();

// <k1,v1>(行偏移量，行值) --> <k2,v2> (行值，1)
    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        data.set(Integer.parseInt(line));
        context.write(data, new IntWritable(1));
    }
}
```

NumberReducer类：

```java
public class NumberReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {

private static IntWritable linenum = new IntWritable(1);

@Override
    protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        // 数据已经按key(行值)进行排序
        // <k2,[v2,v2,v2...]> --> <k2,v2> (linenum，行值)
        for (IntWritable val : values) {
            context.write(linenum, key);
            linenum = new IntWritable(linenum.get() + 1);
        }
    }
}
```

NumberDriver类：

```java
public class NumberDriver {

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
        // file1.txt，file2.txt，file3.txt放入number文件夹中，output文件夹不能存在
        args = new String[]{"E:/develop/skillStudy/tempData/number",
                "E:/develop/skillStudy/output"};
        // 1.获取配置信息以及封装任务
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        // 2.设置jar加载路径
        job.setJarByClass(NumberDriver.class);
        // 3.设置map和reduce类
        job.setMapperClass(NumberMapper.class);
        job.setReducerClass(NumberReducer.class);
        // 4.设置map输出
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        // 5.设置最终输出kv类型
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        // 6.设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        // 7.提交
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}
```

#### 2.使用Hbase相关API操作和协处理器

需求：在社交网站，社交APP上会存储有大量的用户数据以及用户之间的关系数据，比如A用户的好友列表会
展示出他所有的好友，现有一张Hbase表，存储就是当前注册用户的好友关系数据，如下：

![image-20200924010517870](https://gitee.com/itzlg/mypictures/raw/master/img/image-20200924010517870.png)

要求：

- 使用Hbase相关API创建一张结构如上的表。
- 删除好友操作实现（好友关系双向，一方删除好友，另一方也会被迫删除好友）例如：uid1用户执行删除uid2这个好友，则uid2的好友列表中也必须删除uid1。

代码实现

1.引入依赖：

```xml
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>1.3.1</version>
</dependency>
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-server</artifactId>
    <version>1.3.1</version>
</dependency>
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.12</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testng</groupId>
    <artifactId>testng</artifactId>
    <version>6.14.3</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.13</version>
    <scope>compile</scope>
</dependency>
```

HbaseClientTest类：

```java
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

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

public class HbaseClientDemo {

Configuration conf = null;
    Connection conn = null;
    HBaseAdmin admin = null;

@Before
    public void init() throws IOException {
        // 获取一个配置文件对象
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "atzlg6");
//        conf.set("hbase.zookeeper.property.clientPort", "2181");
        // 通过配置对象获取连接
        conn = ConnectionFactory.createConnection(conf);
    }

// 创建一张hbase表
    @Test
    public void createTable() throws IOException {
        admin = (HBaseAdmin) conn.getAdmin();
        //创建表描述器
        HTableDescriptor teacher = new HTableDescriptor(TableName.valueOf("relationship"));
        //设置列族描述器
        teacher.addFamily(new HColumnDescriptor("friends"));
        //执行创建操作
        admin.createTable(teacher);
        System.out.println("relationship表创建成功！！");
    }

//插入多条数据
    @Test
    public void putData() throws IOException {
        //获取一个表对象
        Table t = conn.getTable(TableName.valueOf("relationship"));
        List<Put> putList = new ArrayList<Put>();
        //设定rowkey
        Put put = new Put(Bytes.toBytes("jack"));
        //列族，列名，列值value
        put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("mary"), Bytes.toBytes("mary"));
        put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("tom"), Bytes.toBytes("tom"));
        put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("cita"), Bytes.toBytes("cita"));
        putList.add(put);
        //批量执行插入数据
        t.put(putList);

Put put2 = new Put(Bytes.toBytes("mary"));
        put2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("jack"), Bytes.toBytes("jack"));
        put2.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("tom"), Bytes.toBytes("tom"));
        t.put(put2);
        Put put3 = new Put(Bytes.toBytes("tom"));
        put3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("jack"), Bytes.toBytes("jack"));
        put3.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("cita"), Bytes.toBytes("cita"));
        t.put(put3);
        Put put4 = new Put(Bytes.toBytes("cita"));
        put4.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("jack"), Bytes.toBytes("jack"));
        put4.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("mary"), Bytes.toBytes("mary"));
        t.put(put4);
        //关闭table对象
        t.close();
        System.out.println("插入成功！！");
    }

//删除一条数据
    @Test
    public void deleteData() throws IOException {
        //需要获取一个table对象
        final Table worker = conn.getTable(TableName.valueOf("relationship"));
        //准备delete对象:rowkey
        final Delete delete = new Delete(Bytes.toBytes("cita"));
        //执行删除
        worker.delete(delete);
        //关闭table对象
        worker.close();
        System.out.println("删除数据成功！！");
    }

//查询某个列族数据
    @Test
    public void getDataByCF() throws IOException {
        //获取表对象
        HTable student = (HTable) conn.getTable(TableName.valueOf("relationship"));
        //创建查询的get对象
        Get get = new Get(Bytes.toBytes("jack"));
        //指定列族信息
        // get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"));
        get.addFamily(Bytes.toBytes("friends"));
        //执行查询
        Result res = student.get(get);
        Cell[] cells = res.rawCells();//获取改行的所有cell对象
        for (Cell cell : cells) {
            //通过cell获取rowkey,cf,column,value
            String cf = Bytes.toString(CellUtil.cloneFamily(cell));
            String column = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
            System.out.println(rowkey + "----" + cf + "---" + column + "---" + value);
        }
        student.close();//关闭表对象资源
    }

//通过Scan全表扫描
    @Test
    public void scanAllData() throws IOException {
        HTable student = (HTable) conn.getTable(TableName.valueOf("relationship"));
        Scan scan = new Scan();
        ResultScanner resultScanner = student.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();//获取改行的所有cell对象
            for (Cell cell : cells) {
                //通过cell获取rowkey,cf,column,value
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "----" + cf + "--" + column + "---" + value);
            }
        }
        student.close();
    }

//通过startRowKey和endRowKey进行扫描查询：指定scan的开始rowkey和结束rowkey
    @Test
    public void scanRowKey() throws IOException {
        HTable teacher = (HTable) conn.getTable(TableName.valueOf("relationship"));
        Scan scan = new Scan();
        //指定查询的rowkey区间，rowkey在hbase中以字典序排列
        scan.setStartRow("0001".getBytes());
        scan.setStopRow("2".getBytes());
        ResultScanner resultScanner = teacher.getScanner(scan);
        for (Result result : resultScanner) {
            Cell[] cells = result.rawCells();//获取改行的所有cell对象
            for (Cell cell : cells) {
                //通过cell获取rowkey,cf,column,value
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                System.out.println(rowkey + "----" + cf + "--" + column + "---" + value);
            }
        }
        teacher.close();
    }

// 释放资源
    @After
    public void destroy() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        // 释放连接
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
```

协处理器类：

```java
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

// 继承协处理器BaseRegionObserver
public class DelRelation extends BaseRegionObserver {

// 重写postDelete
    @Override
    public void postDelete(final ObserverContext<RegionCoprocessorEnvironment> e, final Delete delete, final WALEdit edit, final Durability durability) throws IOException {
        super.postDelete(e, delete, edit, durability);
        //获取表对象
        HTableWrapper roles = (HTableWrapper)e.getEnvironment().getTable(TableName.valueOf("relationship"));
        //获取执行删除的列族队列集合
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        //通过队列集合
        Set<Map.Entry<byte[], List<Cell>>> entries = familyCellMap.entrySet();
        //循环队列获取cells对象
        for (final Map.Entry<byte[], List<Cell>> entry : entries) {
            System.out.println(Bytes.toString(entry.getKey()));
            List<Cell> cells = entry.getValue();
            //循环cells对象获取每个cell的rowkey以及列族
            for (final Cell cell : cells) {
                byte[] rowkey = CellUtil.cloneRow(cell);
                byte[] qualifier = CellUtil.cloneQualifier(cell);
                //判断是否存在删除的rowkey的cell
                boolean flag = roles.exists(new Get(qualifier).addColumn(Bytes.toBytes("friends"), rowkey));

if (flag){
                    //生成一个delete对象对需要删除的列族进行删除
                    Delete delQualifier = new Delete(qualifier).addColumn(Bytes.toBytes("friends"), rowkey);
                    roles.delete(delQualifier);
                }

}
        }
    }
}
```

**单机下安装hbase步骤**：

1.修改linux上hosts文件和window上的host文件，添加ip和主机名的映射关系，后进行Linux重启

```
192.168.91.110 atzlg6
```

2.下载安装包：http://archive.apache.org/dist/hbase/1.3.1/  上传到服务器 /opt/lagou/software

3.解压安装包到指定目录：tar -zxvf hbase-1.3.1-bin.tar.gz -C /opt/lagou/servers

4.修改conf目录下配置文件

修改 hbase-env.sh：

```sh
#添加java环境变量
export JAVA_HOME=/opt/module/jdk1.8.0_231
#指定使用外部的zk集群
export HBASE_MANAGES_ZK=true
```

在hbase根目录下创建数据文件夹：mkdir /opt/lagou/servers/hbase-1.3.1/tmpData

修改 hbase-site.xml：

```xml
<configuration>
    <property>
        <name>hbase.rootdir</name>
        <value>file:///opt/lagou/servers/hbase-1.3.1/tmpData</value>
    </property>
</configuration>
```

5.配置Hbase环境变量：vi /etc/profile

```
export HBASE_HOME=/opt/lagou/servers/hbase-1.3.1
export PATH=$PATH:$HBASE_HOME/bin

##JAVA_HOME
export JAVA_HOME=/opt/lagou/servers/jdk1.8.0_231
export CLASSPATH=$CLASSPATH:$JAVA_HOME/lib/
export PATH=$PATH:$JAVA_HOME/bin

##HADOOP_HOME
export HADOOP_HOME=/opt/lagou/servers/hadoop-2.9.2
export PATH=$PATH:$HADOOP_HOME/bin
export PATH=$PATH:$HADOOP_HOME/sbin
```

6.查看当前hbase版本：hbase version；查看正在运行的：jps

7.启动与停止

```shell
cd /opt/lagou/servers/hbase-1.3.1/
# 启动
bin/start-hbase.sh
# 停止
bin/stop-hbase.sh
# 启动hbase shell
bin/hbase shell
```

8. 浏览器访问 [http://192.168.91.110:16010/](http://192.168.1.114:16010/)

**linux上安装hbase后**：

1.打成Jar包上传到服务器

2.单机下挂载协处理器

```shell
describe 'relationship'
alter 'relationship', METHOD => 'table_att','coprocessor' => '/opt/lagou/software/processor.jar|com.fishleap.work.DelRelation|1001'
describe 'relationship'
```

3.验证协处理器：删除表中一条数据(shell方式验证)

```shell
delete 'relationship', 'tom', 'friends:jack'
```

4.卸载协处理器

```shell
disable 'relationship'
alter 'relationship',METHOD=>'table_att_unset',NAME=>'coprocessor$1'
enable 'relationship'
```

效果展示：

添加数据后shell中数据情况：

![image-20200924014056312](https://gitee.com/itzlg/mypictures/raw/master/img/image-20200924014056312.png)

使用shell方式删除一条数据：delete 'relationship', 'mary', 'friends:jack'  后协处理器就会监听到，然后删除 Jack rowkey中 friends:mary 这条数据.

itzlg/phase-four-module-five

简介

发行版

贡献者

近期动态

itzlg/phase-four-module-five .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

itzlg/phase-four-module-five