hbase

README

tRPC-GO hbase 插件

封装 hbase，配合 trpc 使用

hbase client 框架配置

- name: trpc.app.hbase.sevice
  callee: trpc.app.hbase.sevice
  protocol: hbase
  target: hbase://10.0.0.100:2181,10.0.0.101:2181?zookeeperRoot=/root&zookeeperTimeout=1000&regionLookupTimeout=1000&regionReadTimeout=1000&effectiveUser=root

注意事项

gohbase 使用了 logrus 会打一些 debug 日志，hbase.SetLogLevel(log.LevelXXX) 这个用来减少 hbase client 连接过程中的日志

用法

proxy := hbase.NewClientProxy(req.Name) // 请定义为全局变量

// get
table := "tableName"
rowKey := "rowKey"
// 列族->查询列数组
family := make(map[string][]string)
result, err := hbase.ParseGetResult(proxy.Get(ctx, table, rowKey, family))

// put
table := "tableName"
rowKey := "rowKey"
// 列族 -> [列 -> 值]
family := make(map[string]map[string][]byte)
putRsp, err := proxy.Put(ctx, req.Table, req.RowKey, family)

// rawGet -------
family := map[string][]string{
  "family": {"qualifier1", "qualifier2"},
}
get, err := hrpc.NewGetStr(ctx, table, rowKey, hrpc.Families(family))
// judge err
result, err := proxy.RawGet(ctx, get)

// rawPut -------
values := map[string]map[string][]byte{
  "family": {
    "qualifier1": []byte("1"),
    "qualifier2": []byte("str"),
  },
}

put, err := hrpc.NewPutStr(ctx, table, rowKey, values)
// judge err
result, err := proxy.RawPut(ctx, put)

// rawInc -------
inc, err := hrpc.NewIncStrSingle(ctx, table, rowKey, "family", "qualifier1", 2)
// judge err
result, err := proxy.RawInc(ctx, inc)

// del -------
values := map[string]map[string][]byte{
  "family": {
    "qualifier1": nil,
  },
}

// way 1
result, err := proxy.Del(ctx, table, rowKey, values)

// way 2
del, err := hrpc.NewDelStr(ctx, table, rowKey, values)
// judge err
result, err = proxy.RawDel(ctx, del)

Documentation

Overview

Package hbase 封装标准库hbase

Index

• Variables
• func NewClientTransport(opt ...transport.ClientTransportOption) transport.ClientTransport
• func ParseGetResult(result *hrpc.Result, err error) (map[string]map[string]string, error)
• func SetLogLevel(level log.Level)
• type Client
• type ClientCodec
  • func (c *ClientCodec) Decode(msg codec.Msg, buffer []byte) (body []byte, err error)
  • func (c *ClientCodec) Encode(msg codec.Msg, body []byte) (buffer []byte, err error)
• type ClientTransport
  • func (ct *ClientTransport) RoundTrip(ctx context.Context, reqBuf []byte, callOpts ...transport.RoundTripOption) (rspBuf []byte, err error)
• type Config
• type Plugin
  • func (p *Plugin) Setup(name string, configDec plugin.Decoder) error
  • func (p *Plugin) Type() string
• type Request
• type Response

Variables

var (
	DefaultClientCodec = &ClientCodec{}
)

default codec

var DefaultClientTransport = NewClientTransport()

DefaultClientTransport 默认client hbase transport

var ErrRowKeyNotExist = errors.New("ErrRowKeyNotExist")

ErrRowKeyNotExist RowKey不存在

var NewClientProxy = func(name string, opts ...client.Option) Client {
	c := &hbaseCli{
		ServiceName: name,
		Client:      client.DefaultClient,
	}

	c.opts = make([]client.Option, 0, len(opts)+2)
	c.opts = append(c.opts, opts...)
	c.opts = append(c.opts, client.WithProtocol("hbase"), client.WithDisableServiceRouter())

	logrus.SetOutput(defaultlogger)
	return c
}

NewClientProxy 新建一个hbase后端请求代理 必传参数 hbase服务名: trpc.hbase.xxx.xxx

Functions

func NewClientTransport

func NewClientTransport(opt ...transport.ClientTransportOption) transport.ClientTransport

NewClientTransport 创建hbase transport

func ParseGetResult

func ParseGetResult(result *hrpc.Result, err error) (map[string]map[string]string, error)

ParseGetResult 读取hbase内容

func SetLogLevel

func SetLogLevel(level log.Level)

SetLogLevel 设置log leve级别

Types

type Client

type Client interface {
	Get(ctx context.Context, table, rowKey string, family map[string][]string) (*hrpc.Result, error)
	Put(ctx context.Context, table, rowKey string, value map[string]map[string][]byte) (*hrpc.Result, error)
	Del(ctx context.Context, table, rowKey string, value map[string]map[string][]byte) (*hrpc.Result, error)
	RawGet(ctx context.Context, get *hrpc.Get) (*hrpc.Result, error)
	RawPut(ctx context.Context, put *hrpc.Mutate) (*hrpc.Result, error)
	RawDel(ctx context.Context, del *hrpc.Mutate) (*hrpc.Result, error)
	RawScan(ctx context.Context, scan *hrpc.Scan) (hrpc.Scanner, error)
	RawAppend(ctx context.Context, a *hrpc.Mutate) (*hrpc.Result, error)
	RawInc(ctx context.Context, inc *hrpc.Mutate) (int64, error)
	RawCheckAndPut(ctx context.Context, put *hrpc.Mutate,
		family string, qualifier string, expectedValue []byte) (bool, error)
	RawClose(ctx context.Context) error
}

Client HBase查询接口

type ClientCodec

type ClientCodec struct{}

ClientCodec 解码hbase client请求

func (*ClientCodec) Decode

func (c *ClientCodec) Decode(msg codec.Msg, buffer []byte) (body []byte, err error)

Decode 解析hbase client回包里的元数据

func (*ClientCodec) Encode

func (c *ClientCodec) Encode(msg codec.Msg, body []byte) (buffer []byte, err error)

Encode 设置hbase client请求的元数据

type ClientTransport

type ClientTransport struct {
	// contains filtered or unexported fields
}

ClientTransport client端 hbase transport

func (*ClientTransport) RoundTrip

func (ct *ClientTransport) RoundTrip(ctx context.Context, reqBuf []byte,
	callOpts ...transport.RoundTripOption) (rspBuf []byte, err error)

RoundTrip 收发hbase包, 回包hbase response放到ctx里面，这里不需要返回rspBuf

type Config

type Config struct {
	Addr                string `yaml:"addr"`
	ZookeeperRoot       string `yaml:"zookeeperRoot"`
	ZookeeperTimeout    int    `yaml:"zookeeperTimeout"`    // 单位毫秒
	RegionLookupTimeout int    `yaml:"regionLookupTimeout"` // 单位毫秒
	RegionReadTimeout   int    `yaml:"regionReadTimeout"`   // 单位毫秒
	EffectiveUser       string `yaml:"effectiveUser"`       // 访问hbase的用户名
}

Config HBase配置 hbase://zk_addr?zookeeperRoot=/xx/xxx&zookeeperTimeout=100&regionLookupTimeout=100 &regionReadTimeout=100&effectiveUser=root

type Plugin

type Plugin struct{}

Plugin proxy 插件默认初始化, 用于加载hbase相关参数配置

func (*Plugin) Setup

func (p *Plugin) Setup(name string, configDec plugin.Decoder) error

Setup 安装插件

func (*Plugin) Type

func (p *Plugin) Type() string

Type 获取插件类型

type Request

type Request struct {
	Command       string
	Table         string
	RowKey        string
	Family        map[string][]string
	Value         map[string]map[string][]byte
	Get           *hrpc.Get
	Put           *hrpc.Mutate
	Del           *hrpc.Mutate
	Scan          *hrpc.Scan
	Append        *hrpc.Mutate
	Inc           *hrpc.Mutate
	FamilyStr     string
	QualifierStr  string
	ExpectedValue []byte
}

Request hbase request body

type Response

type Response struct {
	Result         *hrpc.Result
	Scanner        hrpc.Scanner
	IncVal         int64
	CheckAndPutVal bool
}

Response hbase response body