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 family103 $t = results;104 echo(column families in {$t}:\n);105 $descriptors = $client->getcolumndescriptors( $t );106 asort( $descriptors );10