php操作hbase例证

php操作hbase例子

1 $globals['thrift_root'] = '/home/thrift';
2 require_once $globals['thrift_root'].'/thrift.php';
3 require_once $globals['thrift_root'].'/protocol/tbinaryprotocol.php';
4 require_once $globals['thrift_root'].'/transport/tsocket.php';
5 require_once $globals['thrift_root'].'/transport/thttpclient.php';
6 require_once $globals['thrift_root'].'/transport/tbufferedtransport.php';
7 require_once $globals['thrift_root'].'/packages/hbase/hbase.php';
8 require_once($globals['thrift_root'].'/packages/hadoopfs/thrifthadoopfilesystem.php');
9
10 $socket = new tsocket(192.168.1.4, 9091);
11 $socket->setsendtimeout(20000);
12 $socket->setrecvtimeout(20000);
13 $transport = new tbufferedtransport($socket);
14 $protocol = new tbinaryprotocol($transport);
15 $hbase = new hbaseclient($protocol);
16 $transport->open();
17
18 // 在这里实现功能
19
20 $transport->close();
21
22 // 多记录批量提交(200提交一次时测试小记录大概在5000/s左右):
23 $rows = array('timestamp'=>$timestamp, 'columns'=>array('txt:col1'=>$col1, 'txt:col2'=>$col2, 'txt:col3'=>$col3));
24 $records = array(rowkey=>$rows,...);
25 $batchrecord = array();
26 foreach ($records as $rowkey => $rows) {
27 $timestamp = $rows['timestamp'];
28 $columns = $rows['columns'];
29 // 生成一条记录
30 $record = array();
31 foreach($columns as $column => $value) {
32 $col = new mutation(array('column'=>$column, 'value'=>$value));
33 array_push($record, $col);
34 }
35 // 加入记录数组
36 $batchtmp = new batchmutation(array('row'=>$rowkey, 'mutations'=>$record));
37 array_push($batchrecord, $batchtmp);
38 }
39 $ret = $hbase->mutaterows('test', $batchrecord);
40
41 // 单记录提交(1000/s左右)
42 $mutation = array(new mutation(array('column'=>'txt:col1', 'value'=>$col1)),
43 new mutation(array('column'=>'txt:col2', 'value'=>$col2)),
44 new mutation(array('column'=>'txt:col3', 'value'=>$col3)));
45 $hbase->mutaterow('test', $rowkey, $mutation);
46
47 // 扫描记录
48 $result = $hbase->scanneropenwithstop($table, $startkey, $endkey, $columns);
49 while (true) {
50 $record = $hbase->scannerget($result);
51 if ($record == null) {
52 break;
53 }
54 $recordarray = array();
55 foreach($record as $trowresult) {
56 $row = $trowresult->row;
57 $column = $trowresult->columns;
58 foreach($column as $family_column=>$cell) {
59 $recordarray[$family_column] = $cellval;
60 }
61 $resultarray[] = $recordarray;
62 }
63 }
64 print_r($resultarray);
65
66 // 以下记录从别人处(佛祖球球)拷贝,把几个接口补全,(未测试...)
67
68 // 列出hbase 裡的所有 table
69 echo( "listing tables...\n" );
70 $tables = $client->gettablenames();
71 sort( $tables );
72 foreach ( $tables as $name )
73 {
74 echo $name."\n";
75 }
76
77 // 刪除table
78 $name = "test2";
79 if($client->istableenabled($name))
80 {
81 echo "关闭".$name."资料表\n";
82 $client->disabletable($name);
83 }
84 echo "刪除中...\n";
85 $client->deletetable($name);
86 echo "刪除完成";
87
88 // 新增table
89 $columns = array(new columndescriptor(array('name' => 'name:')),
90 new columndescriptor(array('name'=> 'scores:',)));
91 $t = "test2";
92 echo("creating table: {$t}\n");
93 try
94 {
95 $client->createtable( $t, $columns );
96 }
97 catch (alreadyexists $ae)
98 {
99 echo( "warn: {$ae->message}\n" );
100 }
101
102 //列出table內的column family
103 $t = "results";
104 echo("column families in {$t}:\n");
105 $descriptors = $client->getcolumndescriptors( $t );
106 asort( $descriptors );
10