< 文档首页
全部产品
弹性计算
存储
数据库
网络
视频与CDN
域名与网站(万网)
应用服务
互联网中间件
移动研发平台EMAS
移动研发平台EMAS-专有云
云通信
安全
大数据
人工智能
ET大脑
物联网
物联网行业方案
数字金融
管理与监控
云市场
开发者工具
解决方案
智能硬件
会员服务
更多

    多语言(C++/Python/Go等)API访问

    更新时间：2019-10-15 14:49:50

    ★ 我的收藏
    本页目录

    简介

    HBase增强版通过Thrift支持多语言访问，只要是Thrift支持的语言，都可以访问HBase增强版。HBase增强版服务端的Thrift版本是0.12.0，虽然说thrift支持向后兼容，但如果有条件的用户，最好还是下载0.12.0的thrift，点击这里下载。一些语言提供了管理依赖的方法，可以遵循这些语言的使用习惯来安装thrift，如Python语言可以通过pip install thrift来安装，Go语言可以直接在代码里import {"github.com/apache/thrift/lib/go/thrift"}即可。

    HBase增强版使用的Thrift接口定义是HBase的thrift2。因此需要下载thrift2的定义文件生成相应语言的接口。相比thrift1定义，HBase中的thrift2接口定义更加清晰，用户可以获得和Java语言类似的API调用体验。之前的thrift2之所以使用不广泛是因为thrift2的接口不全，不支持建表删表等DDL操作。但是目前thrift2相关的API定义已经由阿里云的开发人员补全并回馈给了社区（HBASE-21649）。目前thrift2比thrift1的功能更全而且更容易使用。

    准备

    1. 下载Thrift安装包，点击这里下载。
    2. 下载HBase Thrift2定义文件
    3. 获取集群的连接地址,多语言访问使用的连接为连接信息中的“非 JAVA 语言 Thrift2 访问地址”访问地址部分（注意专有网络地址和外网地址的区别）。

    访问

    Thrift的使用帮助可以参考Apache Thrift的官方文档,下面给出简单的使用方法。

    1.生成对应语言的接口定义文件

    从上述地址下载接口定义文件后，按照如下语法生成接口定义文件

    1.  thrift --gen <language> Hbase.thrift

    例如：

    1. thrift --gen php Hbase.thrift
    2. thrift --gen cpp Hbase.thrift
    3. thrift --gen py Hbase.thrift

    2. 构造客户端访问HBase增强版

    HBase增强版Thrift服务器端的transport层使用的是HTTP，因此在构造客户端时，需要thrift中的ThttpClient（各个语言都有相应实现）。并且在ACL开启的情况下，需要在ThttpClient上加上两个header来向服务器传输用户名和密码进行认证（如果关闭ACL则不需要）。Thrift在每个语言实现的ThttpClient都有加定制header的函数。

    python示例

    1. # -*- coding: utf-8  -*-
    2. # 以下两个模块可以通过 pip install thrift 安装获得
    3. from thrift.protocol import TBinaryProtocol
    4. from thrift.transport import THttpClient
    6. # 下面的模块通过 thrift --gen py hbase.thrift 来生成
    7. from hbase import THBaseService
    8. from hbase.ttypes import TColumnValue, TColumn, TTableName, TTableDescriptor, TColumnFamilyDescriptor, TNamespaceDescriptor, TGet, TPut, TScan
    10. # 连接地址
    11. url = "http://host:9190"
    12. transport = THttpClient.THttpClient(url)
    13. headers = {}
    14. # 用户名
    15. headers["ACCESSKEYID"]="root";
    16. # 密码
    17. headers["ACCESSSIGNATURE"]="root"
    18. transport.setCustomHeaders(headers)
    19. protocol = TBinaryProtocol.TBinaryProtocolAccelerated(transport)
    20. client = THBaseService.Client(protocol)
    21. transport.open()
    23. # create Namespace
    24. client.createNamespace(TNamespaceDescriptor(name="ns"))
    26. # create table
    27. tableName = TTableName(ns="ns", qualifier="table1")
    28. client.createTable(TTableDescriptor(tableName=tableName, columns=[
    29.     TColumnFamilyDescriptor(name="f")
    30. ]), None)
    33. # 做DML操作时，表名参数为bytes，表名的规则是namespace + 冒号 + 表名
    34. tableInbytes = "ns:table1".encode("utf8")
    36. # 插入数据
    37. client.put(table=tableInbytes, tput=TPut(row="row1".encode("utf8"), columnValues=[TColumnValue(family="f".encode("utf8"), qualifier="q1".encode("utf8"), value="value".encode("utf8"))]))
    39. # 批量插入数据
    40. puts = []
    41. put1 = TPut(row="row2".encode("utf8"), columnValues=[TColumnValue(family="f".encode("utf8"), qualifier="q1".encode("utf8"), value="value2".encode("utf8"))])
    42. put2 = TPut(row="row3".encode("utf8"), columnValues=[TColumnValue(family="f".encode("utf8"), qualifier="q1".encode("utf8"), value="value3".encode("utf8"))])
    43. puts.append(put1)
    44. puts.append(put2)
    45. client.putMultiple(table=tableInbytes, tputs=puts)
    48. # 单行查询数据
    49. result = client.get(tableInbytes, TGet(row="row1".encode("utf8")))
    50. print "Get result: ", result
    52. # 批量单行查询
    53. gets = []
    54. get1 = TGet(row="row2".encode("utf8"))
    55. get2 = TGet(row="row3".encode("utf8"))
    56. gets.append(get1)
    57. gets.append(get2)
    58. results = client.getMultiple(tableInbytes, gets)
    60. print "getMultiple results: ", results
    62. # 范围扫描
    63. # 扫描时需要设置startRow和stopRow，否则会变成全表扫描
    64. startRow = "row0".encode("utf8")
    65. stopRow = "row9".encode("utf8")
    66. scan = TScan(startRow=startRow, stopRow=stopRow)
    67. # caching的大小为每次从服务器返回的行数，设置太大会导致服务器处理过久，太小会导致范围扫描与服务器做过多交互
    68. # 根据每行的大小，caching的值一般设置为10到100之间
    69. caching = 2
    71. # 扫描的结果
    72. results = []
    74. # 此函数可以找到比当前row大的最小row，方法是在当前row后加入一个0x00的byte
    75. # 从比当前row大的最小row开始scan，可以保证中间不会漏扫描数据
    76. def createClosestRowAfter(row):
    77.     array = bytearray(row)
    78.     array.append(0x00)
    79.     return bytes(array)
    81. while True:
    82.     lastResult = None
    83.     # getScannerResults会自动完成open,close 等scanner操作，HBase增强版必须使用此方法进行范围扫描
    84.     currentResults = client.getScannerResults(tableInbytes, scan, caching)
    85.     for result in currentResults:
    86.         results.append(result)
    87.         lastResult = result
    88.     # 如果一行都没有扫描出来，说明扫描已经结束，我们已经获得startRow和stopRow之间所有的result    
    89.     if lastResult is None:
    90.         break
    91.     # 如果此次扫描是有结果的，我们必须构造一个比当前最后一个result的行大的最小row，继续进行扫描，以便返回所有结果
    92.     else:
    93.         nextStartRow = createClosestRowAfter(lastResult.row)
    94.         scan = TScan(startRow=nextStartRow, stopRow=stopRow)
    95. print "Scan results: ", results
    97. # disable 表
    98. client.disableTable(TTableName(ns="ns", qualifier="table1"))
    100. # 删除表
    101. client.deleteTable(TTableName(ns="ns", qualifier="table1"))
    103. # close 连接
    104. transport.close()

    Go 语言示例

    1. package main
    3. import (
    4.     // hbase模块通过 thrift --gen go hbase.thrift 来生成
    5.     "./hbase"
    6.     "context"
    7.     "fmt"
    8.     "github.com/apache/thrift/lib/go/thrift"
    9.     "os"
    10. )
    12. const (
    13.     HOST = "http://host:9190"
    14.     // 用户名
    15.     USER = "root"
    16.     // 密码
    17.     PASSWORD = "root"
    18. )
    20. func main() {
    21.     defaultCtx := context.Background()
    22.     protocolFactory := thrift.NewTBinaryProtocolFactoryDefault()
    23.     trans, err := thrift.NewTHttpClient(HOST)
    24.     if err != nil {
    25.         fmt.Fprintln(os.Stderr, "error resolving address:", err)
    26.         os.Exit(1)
    27.     }
    28.     // 设置用户名密码
    29.     httClient := trans.(*thrift.THttpClient)
    30.     httClient.SetHeader("ACCESSKEYID", USER)
    31.     httClient.SetHeader("ACCESSSIGNATURE", PASSWORD)
    32.     client := hbase.NewTHBaseServiceClientFactory(trans, protocolFactory)
    33.     if err := trans.Open(); err != nil {
    34.         fmt.Fprintln(os.Stderr, "Error opening "+HOST, err)
    35.         os.Exit(1)
    36.     }
    38.     // create Namespace
    39.     err = client.CreateNamespace(defaultCtx, &hbase.TNamespaceDescriptor{Name: "ns"})
    40.     if err != nil {
    41.         fmt.Fprintln(os.Stderr, "error CreateNamespace:", err)
    42.         os.Exit(1)
    43.     }
    45.     // create table
    46.     tableName := hbase.TTableName{Ns: []byte("ns"), Qualifier: []byte("table1")}
    47.     err = client.CreateTable(defaultCtx, &hbase.TTableDescriptor{TableName: &tableName, Columns: []*hbase.TColumnFamilyDescriptor{&hbase.TColumnFamilyDescriptor{Name: []byte("f")}}}, nil)
    48.     if err != nil {
    49.         fmt.Fprintln(os.Stderr, "error CreateTable:", err)
    50.         os.Exit(1)
    51.     }
    53.     //做DML操作时，表名参数为bytes，表名的规则是namespace + 冒号 + 表名
    54.     tableInbytes := []byte("ns:table1")
    56.     // 插入数据
    57.     err = client.Put(defaultCtx, tableInbytes, &hbase.TPut{Row: []byte("row1"), ColumnValues: []*hbase.TColumnValue{&hbase.TColumnValue{
    58.         Family:    []byte("f"),
    59.         Qualifier: []byte("q1"),
    60.         Value:     []byte("value1")}}})
    61.     if err != nil {
    62.         fmt.Fprintln(os.Stderr, "error Put:", err)
    63.         os.Exit(1)
    64.     }
    66.     //批量插入数据
    67.     puts := []*hbase.TPut{&hbase.TPut{Row: []byte("row2"), ColumnValues: []*hbase.TColumnValue{&hbase.TColumnValue{
    68.         Family:    []byte("f"),
    69.         Qualifier: []byte("q1"),
    70.         Value:     []byte("value2")}}}, &hbase.TPut{Row: []byte("row3"), ColumnValues: []*hbase.TColumnValue{&hbase.TColumnValue{
    71.         Family:    []byte("f"),
    72.         Qualifier: []byte("q1"),
    73.         Value:     []byte("value3")}}}}
    74.     err = client.PutMultiple(defaultCtx, tableInbytes, puts)
    75.     if err != nil {
    76.         fmt.Fprintln(os.Stderr, "error PutMultiple:", err)
    77.         os.Exit(1)
    78.     }
    80.     // 单行查询数据
    81.     result, err := client.Get(defaultCtx, tableInbytes, &hbase.TGet{Row: []byte("row1")})
    82.     fmt.Println("Get result:")
    83.     fmt.Println(result)
    85.     // 批量单行查询
    86.     gets := []*hbase.TGet{&hbase.TGet{Row: []byte("row2")}, &hbase.TGet{Row: []byte("row3")}}
    88.     results, err := client.GetMultiple(defaultCtx, tableInbytes, gets)
    89.     fmt.Println("GetMultiple result:")
    90.     fmt.Println(results)
    92.     //范围扫描
    94.     //扫描时需要设置startRow和stopRow，否则会变成全表扫描
    95.     startRow := []byte("row0")
    96.     stopRow := []byte("row9")
    98.     scan := &hbase.TScan{StartRow: startRow, StopRow: stopRow}
    99.     //caching的大小为每次从服务器返回的行数，设置太大会导致服务器处理过久，太小会导致范围扫描与服务器做过多交互
    100.     //根据每行的大小，caching的值一般设置为10到100之间
    101.     caching := 2
    103.     // 扫描的结果
    104.     var scanResults []*hbase.TResult_
    106.     for true {
    107.         var lastResult *hbase.TResult_ = nil
    108.         // getScannerResults会自动完成open,close 等scanner操作，HBase增强版必须使用此方法进行范围扫描
    109.         currentResults, _ := client.GetScannerResults(defaultCtx, tableInbytes, scan, int32(caching))
    111.         for _, tResult := range currentResults {
    112.             lastResult = tResult
    113.             scanResults = append(scanResults, tResult)
    114.         }
    115.         // 如果一行都没有扫描出来，说明扫描已经结束，我们已经获得startRow和stopRow之间所有的result
    116.         if lastResult == nil {
    117.             break
    118.         } else {
    119.             // 如果此次扫描是有结果的，我们必须构造一个比当前最后一个result的行大的最小row，继续进行扫描，以便返回所有结果
    120.             nextStartRow := createClosestRowAfter(lastResult.Row)
    121.             scan = &hbase.TScan{StartRow: nextStartRow, StopRow: stopRow}
    122.         }
    123.     }
    124.     fmt.Println("Scan result:")
    125.     fmt.Println(scanResults)
    127.     //disable table
    128.     err = client.DisableTable(defaultCtx, &tableName)
    129.     if err != nil {
    130.         fmt.Fprintln(os.Stderr, "error DisableTable:", err)
    131.         os.Exit(1)
    132.     }
    134.     // delete table
    135.     err = client.DeleteTable(defaultCtx, &tableName)
    136.     if err != nil {
    137.         fmt.Fprintln(os.Stderr, "error DisableTable:", err)
    138.         os.Exit(1)
    139.     }
    141.     // delete namespace
    142.     err = client.DeleteNamespace(defaultCtx, "ns")
    143.     if err != nil {
    144.         fmt.Fprintln(os.Stderr, "error DisableTable:", err)
    145.         os.Exit(1)
    146.     }
    147.     //tableName, err := client.GetTableNamesByNamespace(defaultCtx, "default")
    148.     if err != nil {
    149.         fmt.Fprintln(os.Stderr, "error getting table:", err)
    150.         os.Exit(1)
    151.     }
    152.     //s := string(tableName[0].Qualifier)
    153.     //fmt.Println(s)
    154. }
    156. // 此函数可以找到比当前row大的最小row，方法是在当前row后加入一个0x00的byte
    157. // 从比当前row大的最小row开始scan，可以保证中间不会漏扫描数据
    158. func createClosestRowAfter(row []byte) []byte {
    159.     var nextRow []byte
    160.     var i int
    161.     for i = 0; i < len(row); i++ {
    162.         nextRow = append(nextRow, row[i])
    163.     }
    164.     nextRow = append(nextRow, 0x00)
    165.     return nextRow
    166. }

    php 示例

    1. //设置用户名和密码，ACCESSKEYID的值为用户名，ACCESSSIGNATURE的值为密码
    2. $headers = array('ACCESSKEYID' => 'root','ACCESSSIGNATURE' => 'root');
    3. //新建HTTP协议的客户端
    4. $socket = new THttpClient('hb-xxxx.aliyuncs.com', 9190);
    5. //将header设置到http client中，访问时携带用户名密码
    6. $socket->addHeaders（$headers）
    7. $transport = new TBufferedTransport($socket);
    8. $protocol = new TBinaryProtocol($transport);
    9. $client = new HbaseClient($protocol);
    10. $transport->open();

    更多语言的使用请参见Thrfit官方文档

    上一篇：全文索引服务(Solr)
    下一篇：Spark访问
    相关文档
    相关产品
    • 云数据库 HBase 版
      云数据库 HBase 版（ApsaraDB for HBase）是基于 Hadoop 的一个分布式数据库，支持海量的PB级的大数据存储，适用于高吞吐的随机读写的场景。目前在阿里内部有数百个集群，10000台左右规模的集群，服务数百个业务线，在订单存储、消息存储、物联网、轨迹、wifi、安全风控、搜索等领域有较多的在线应用。 阿里云特别提供HBase产品化方案服务广大的中小型客户。
    • 时序时空数据库
      时序时空数据库产品系列是是广泛应用于物联网（IoT）设备监控系统 ，企业能源管理系统（EMS），生产安全监控系统，电力检测系统等行业场景的专业数据库产品，提供百万高效写入，高压缩比低成本存储、预降采样、插值、多维聚合计算，查询结果可视化功能；解决由于设备采集点数量巨大，数据采集频率高，造成的存储成本高，写入和查询分析效率低的问题。目前提供：时序数据库 TSDB ，时序数据库 InfluxDB®，时序数据库 Promethues® 和时空数据库 4个产品类型。
    • E-MapReduce
      阿里云 Elastic MapReduce（E-MapReduce） 是一种大数据处理的系统解决方案。构建于阿里云云服务器 ECS 上，基于开源的 Apache Hadoop 和 Apache Spark，让用户可以方便地使用 Hadoop 和 Spark 生态系统中的其他周边系统（如 Apache Hive、Apache Pig、HBase 等）来分析和处理自己的数据。用户还可以通过 E-MapReduce 将数据非常方便的导入和导出到阿里云其他的云数据存储系统和数据库系统中，如阿里云 OSS、阿里云 RDS 等。