Phoenix之协处理器

淡淡的烟草味﹌ 2022-10-25 12:48 219阅读 0赞

# Phoenix之协处理器 #

phoneix的本质就是定义了大量的协处理器，使用协处理器帮助我们完成HBase的操作！

## 1.简介 ##

在RDMS中，可以定义

function（函数)： 在关系型数据库中，必须使用mysql | oracle提供的语法，来定义一个函数！

trigger(触发器): 在关系型数据库中，可以定义触发器，触发器会在执行写操作(insert,update,delete)时，触发触发器的代码，完成一段功能！

procedure（存储过程）： 在关系型数据库中，可以定义存储过程，存储过程和函数的定义是一致的！唯一的区别是函数必须有返回值，而存储过程不能有返回值！理解存储过程是void的函数！

存储过程，必须使用mysql提供的语法，编写一段程序！

需要手动调用 call 存储过程！

存储过程可以完成一段高效的功能！

## 2.协处理器(corprocessor) ##

分两种：

**Observer(观察者)：** 类比为mysql的触发器！ 其中包含了大量的 hook程序(prePut,postPut,preDelete,postDelete)！可以在对hbase中的表进行 put/delete/get/scan 操作的之前和之后被regionserver固定调用！  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

**EndPoint(终端)**: 类比为mysql的存储过程! 是一段在regionserver端计算的程序，可以完成某些在regionserver端就可以完成的运算！ 例如实现对表中数据的Max,Avg等高效统计！  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

## 3.自定义Observer ##

package com.hbase.use;
    
    
    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.ObserverContext;
    import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
    import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
    import org.apache.hadoop.hbase.coprocessor.RegionObserver;
    import org.apache.hadoop.hbase.util.Bytes;
    import org.apache.hadoop.hbase.wal.WALEdit;
    
    import java.io.IOException;
    import java.util.List;
    import java.util.Optional;
    
    public class MyObserver implements RegionObserver, RegionCoprocessor { 
    
        private static Connection conn;
    
        static { 
            try { 
                conn = ConnectionUtil.connection();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    
        @Override
        public Optional<RegionObserver> getRegionObserver() { 
            return Optional.of(this);
        }
    
        //put之前的操作 向t1表插入数据，判断如果数据中的info:age列的值为20，就将这条数据也插入到new表
        @Override
        public void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException { 
    
            //获取t2表的Table对象
            Table table = conn.getTable(TableName.valueOf(null, "new"));
    
            //从当前的put对象中获取info：age的值，进行判断
            List<Cell> cells = put.get(Bytes.toBytes("info"), Bytes.toBytes("age"));
    
            for (Cell cell : cells) { 
                if (Bytes.toString(CellUtil.cloneValue(cell)).equals("20")){ 
                    //将当前put，写入到指定表
                    table.put(put);
                    //跳出循环
                    break;
                }
            }
    
            table.close();
    
        }
    
    
        //put之后的操作 向result表中插入记录，统计某个rowkey历史一共被操作了多少次
        @Override
        public void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException { 
    
            Table table = conn.getTable(TableName.valueOf(null, "audults"));
    
            //从当前put中获取rowkey
            byte[] row = put.getRow();
    
            //从audults表中获取这个rowkey当前的值，如果这个rowkey不存在 插入
            Result result = table.get(new Get(row));
    
            Put newput = new Put(row);
    
            if (result.isEmpty()){ 
                newput.addColumn(Bytes.toBytes("info"),Bytes.toBytes("times"),
                        Bytes.toBytes("1"));
            }else { 
    
                //如果rowkey存在，更新rowkey次数
                Cell cell = result.getColumnLatestCell(Bytes.toBytes("info"), Bytes.toBytes("times"));
    
                //int newvalue = Bytes.toInt(CellUtil.cloneValue(cell)) + 1;
    
                int newvalue = Integer.valueOf(Bytes.toString(CellUtil.cloneValue(cell))) + 1;
    
    
                newput.addColumn(Bytes.toBytes("info"),Bytes.toBytes("times"),
                        Bytes.toBytes(newvalue+""));
    
    
            }
    
            //插入表中
            table.put(newput);
    
            table.close();
    
        }
    }

## 4.测试 ##

将上述类打包上传到hdfs中  
在hbase客户端使用alter命令对表定义协处理器，指定协处理器路径及全类名，即可使用。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221024/4e46bde6cff5439290b7a5205f2b003d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221024/c39f883dfba248b1a813833798ca4759.png